pezkuwichain/revive-differential-tests
1
0
Fork 0
mirror of https://github.com/pezkuwichain/revive-differential-tests.git synced 2026-04-30 07:07:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

Core Benchmarking Infra (#175)

Browse Source 
* Implement a solution for the pre-fund account limit

* Update the account pre-funding handling

* Fix the lighthouse node tracing issue

* refactor existing dt infra

* Implement the platform driver

* Wire up the cleaned up driver implementation

* Implement the core benchmarking components

* Remove some debug logging

* Fix issues in the benchmarks driver

* Implement a global concurrency limit on provider requests

* Update the concurrency limit

* Update the concurrency limit

* Cleanups

* Update the lighthouse ports

* Ignore certain tests

* Update the new geth test
This commit is contained in:
Omar
2025-10-05 18:09:01 +03:00
committed by GitHub
parent f9dc362c03
commit 74fdeb4a2e
51 changed files with 4308 additions and 1990 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/node/src/constants.rs
+5
View File
@@ -1,5 +1,10 @@
use alloy::primitives::ChainId;
/// This constant defines how much Wei accounts are pre-seeded with in genesis.
///
/// Note: After changing this number, check that the tests for substrate work as we encountered
/// some issues with different values of the initial balance on substrate.
pub const INITIAL_BALANCE: u128 = 10u128.pow(37);
/// The chain id used for all of the chains spawned by the framework.
pub const CHAIN_ID: ChainId = 420420420;
crates/node/src/helpers/mod.rs
+3
View File
@@ -0,0 +1,3 @@
mod process;
pub use process::*;
crates/node/src/process.rs → crates/node/src/helpers/process.rs
+7 -4
View File
@@ -110,8 +110,11 @@ impl Process {
}
let check_result =
check_function(stdout_line.as_deref(), stderr_line.as_deref())
.context("Failed to wait for the process to be ready")?;
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;
@@ -127,10 +130,10 @@ impl Process {
ProcessReadinessWaitBehavior::WaitForCommandToExit => {
if !child
.wait()
.context("Failed waiting for kurtosis run process to finish")?
.context("Failed waiting for process to finish")?
.success()
{
anyhow::bail!("Failed to initialize kurtosis network",);
anyhow::bail!("Failed to spawn command");
}
}
}
crates/node/src/lib.rs
+3 -5
View File
@@ -3,12 +3,10 @@
use alloy::genesis::Genesis;
use revive_dt_node_interaction::EthereumNode;
pub mod common;
pub mod constants;
pub mod geth;
pub mod lighthouse_geth;
pub mod process;
pub mod substrate;
pub mod helpers;
pub mod node_implementations;
pub mod provider_utils;
/// An abstract interface for testing nodes.
pub trait Node: EthereumNode {
crates/node/src/geth.rs → crates/node/src/node_implementations/geth.rs
+168 -49
View File
@@ -20,18 +20,22 @@ use alloy::{
network::{Ethereum, EthereumWallet, NetworkWallet},
primitives::{Address, BlockHash, BlockNumber, BlockTimestamp, StorageKey, TxHash, U256},
providers::{
Provider, ProviderBuilder,
Provider,
ext::DebugApi,
fillers::{CachedNonceManager, ChainIdFiller, FillProvider, NonceFiller, TxFiller},
fillers::{CachedNonceManager, ChainIdFiller, NonceFiller},
},
rpc::types::{
EIP1186AccountProofResponse, TransactionRequest,
trace::geth::{DiffMode, GethDebugTracingOptions, PreStateConfig, PreStateFrame},
EIP1186AccountProofResponse, TransactionReceipt, TransactionRequest,
trace::geth::{
DiffMode, GethDebugTracingOptions, GethTrace, PreStateConfig, PreStateFrame,
},
},
};
use anyhow::Context as _;
use futures::{Stream, StreamExt};
use revive_common::EVMVersion;
use tracing::{Instrument, instrument};
use tokio::sync::OnceCell;
use tracing::{Instrument, error, instrument};
use revive_dt_common::{
fs::clear_directory,
@@ -39,13 +43,13 @@ use revive_dt_common::{
};
use revive_dt_config::*;
use revive_dt_format::traits::ResolverApi;
use revive_dt_node_interaction::EthereumNode;
use revive_dt_node_interaction::{EthereumNode, MinedBlockInformation};
use crate::{
Node,
common::FallbackGasFiller,
constants::INITIAL_BALANCE,
process::{Process, ProcessReadinessWaitBehavior},
constants::{CHAIN_ID, INITIAL_BALANCE},
helpers::{Process, ProcessReadinessWaitBehavior},
provider_utils::{ConcreteProvider, FallbackGasFiller, construct_concurrency_limited_provider},
};
static NODE_COUNT: AtomicU32 = AtomicU32::new(0);
@@ -70,7 +74,7 @@ pub struct GethNode {
start_timeout: Duration,
wallet: Arc<EthereumWallet>,
nonce_manager: CachedNonceManager,
chain_id_filler: ChainIdFiller,
provider: OnceCell<ConcreteProvider<Ethereum, Arc<EthereumWallet>>>,
}
impl GethNode {
@@ -119,8 +123,8 @@ impl GethNode {
handle: None,
start_timeout: geth_configuration.start_timeout_ms,
wallet: wallet.clone(),
chain_id_filler: Default::default(),
nonce_manager: Default::default(),
provider: Default::default(),
}
}
@@ -235,7 +239,7 @@ impl GethNode {
match process {
Ok(process) => self.handle = Some(process),
Err(err) => {
tracing::error!(?err, "Failed to start geth, shutting down gracefully");
error!(?err, "Failed to start geth, shutting down gracefully");
self.shutdown()
.context("Failed to gracefully shutdown after geth start error")?;
return Err(err);
@@ -245,27 +249,29 @@ impl GethNode {
Ok(self)
}
async fn provider(
&self,
) -> anyhow::Result<FillProvider<impl TxFiller<Ethereum>, impl Provider<Ethereum>, Ethereum>>
{
ProviderBuilder::new()
.disable_recommended_fillers()
.filler(FallbackGasFiller::new(
25_000_000,
1_000_000_000,
1_000_000_000,
))
.filler(self.chain_id_filler.clone())
.filler(NonceFiller::new(self.nonce_manager.clone()))
.wallet(self.wallet.clone())
.connect(&self.connection_string)
async fn provider(&self) -> anyhow::Result<ConcreteProvider<Ethereum, Arc<EthereumWallet>>> {
self.provider
.get_or_try_init(|| async move {
construct_concurrency_limited_provider::<Ethereum, _>(
self.connection_string.as_str(),
FallbackGasFiller::default(),
ChainIdFiller::new(Some(CHAIN_ID)),
NonceFiller::new(self.nonce_manager.clone()),
self.wallet.clone(),
)
.await
.context("Failed to construct the provider")
})
.await
.map_err(Into::into)
.cloned()
}
}
impl EthereumNode for GethNode {
fn pre_transactions(&mut self) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + '_>> {
Box::pin(async move { Ok(()) })
}
fn id(&self) -> usize {
self.id as _
}
@@ -274,6 +280,50 @@ impl EthereumNode for GethNode {
&self.connection_string
}
#[instrument(
level = "info",
skip_all,
fields(geth_node_id = self.id, connection_string = self.connection_string),
err,
)]
fn submit_transaction(
&self,
transaction: TransactionRequest,
) -> Pin<Box<dyn Future<Output = anyhow::Result<TxHash>> + '_>> {
Box::pin(async move {
let provider = self
.provider()
.await
.context("Failed to create the provider for transaction submission")?;
let pending_transaction = provider
.send_transaction(transaction)
.await
.context("Failed to submit the transaction through the provider")?;
Ok(*pending_transaction.tx_hash())
})
}
#[instrument(
level = "info",
skip_all,
fields(geth_node_id = self.id, connection_string = self.connection_string),
err,
)]
fn get_receipt(
&self,
tx_hash: TxHash,
) -> Pin<Box<dyn Future<Output = anyhow::Result<TransactionReceipt>> + '_>> {
Box::pin(async move {
self.provider()
.await
.context("Failed to create provider for getting the receipt")?
.get_transaction_receipt(tx_hash)
.await
.context("Failed to get the receipt of the transaction")?
.context("Failed to get the receipt of the transaction")
})
}
#[instrument(
level = "info",
skip_all,
@@ -283,8 +333,7 @@ impl EthereumNode for GethNode {
fn execute_transaction(
&self,
transaction: TransactionRequest,
) -> Pin<Box<dyn Future<Output = anyhow::Result<alloy::rpc::types::TransactionReceipt>> + '_>>
{
) -> Pin<Box<dyn Future<Output = anyhow::Result<TransactionReceipt>> + '_>> {
Box::pin(async move {
let provider = self
.provider()
@@ -292,12 +341,12 @@ impl EthereumNode for GethNode {
.context("Failed to create provider for transaction submission")?;
let pending_transaction = provider
.send_transaction(transaction)
.await
.inspect_err(
|err| tracing::error!(%err, "Encountered an error when submitting the transaction"),
)
.context("Failed to submit transaction to geth node")?;
.send_transaction(transaction)
.await
.inspect_err(
|err| error!(%err, "Encountered an error when submitting the transaction"),
)
.context("Failed to submit transaction to geth node")?;
let transaction_hash = *pending_transaction.tx_hash();
// The following is a fix for the "transaction indexing is in progress" error that we used
@@ -317,7 +366,6 @@ impl EthereumNode for GethNode {
// allowed for 60 seconds of waiting with a 1 second delay in polling, we need to allow for
// a larger wait time. Therefore, in here we allow for 5 minutes of waiting with exponential
// backoff each time we attempt to get the receipt and find that it's not available.
let provider = Arc::new(provider);
poll(
Self::RECEIPT_POLLING_DURATION,
PollingWaitBehavior::Constant(Duration::from_millis(200)),
@@ -351,14 +399,12 @@ impl EthereumNode for GethNode {
&self,
tx_hash: TxHash,
trace_options: GethDebugTracingOptions,
) -> Pin<Box<dyn Future<Output = anyhow::Result<alloy::rpc::types::trace::geth::GethTrace>> + '_>>
{
) -> Pin<Box<dyn Future<Output = anyhow::Result<GethTrace>> + '_>> {
Box::pin(async move {
let provider = Arc::new(
self.provider()
.await
.context("Failed to create provider for tracing")?,
);
let provider = self
.provider()
.await
.context("Failed to create provider for tracing")?;
poll(
Self::TRACE_POLLING_DURATION,
PollingWaitBehavior::Constant(Duration::from_millis(200)),
@@ -456,14 +502,54 @@ impl EthereumNode for GethNode {
fn evm_version(&self) -> EVMVersion {
EVMVersion::Cancun
}
fn subscribe_to_full_blocks_information(
&self,
) -> Pin<
Box<
dyn Future<Output = anyhow::Result<Pin<Box<dyn Stream<Item = MinedBlockInformation>>>>>
+ '_,
>,
> {
Box::pin(async move {
let provider = self
.provider()
.await
.context("Failed to create the provider for block subscription")?;
let block_subscription = provider.subscribe_full_blocks();
let block_stream = block_subscription
.into_stream()
.await
.context("Failed to create the block stream")?;
let mined_block_information_stream = block_stream.filter_map(|block| async {
let block = block.ok()?;
Some(MinedBlockInformation {
block_number: block.number(),
block_timestamp: block.header.timestamp,
mined_gas: block.header.gas_used as _,
block_gas_limit: block.header.gas_limit as _,
transaction_hashes: block
.transactions
.into_hashes()
.as_hashes()
.expect("Must be hashes")
.to_vec(),
})
});
Ok(Box::pin(mined_block_information_stream)
as Pin<Box<dyn Stream<Item = MinedBlockInformation>>>)
})
}
}
pub struct GethNodeResolver<F: TxFiller<Ethereum>, P: Provider<Ethereum>> {
pub struct GethNodeResolver {
id: u32,
provider: FillProvider<F, P, Ethereum>,
provider: ConcreteProvider<Ethereum, Arc<EthereumWallet>>,
}
impl<F: TxFiller<Ethereum>, P: Provider<Ethereum>> ResolverApi for GethNodeResolver<F, P> {
impl ResolverApi for GethNodeResolver {
#[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
fn chain_id(
&self,
@@ -648,12 +734,38 @@ mod tests {
(context, node)
}
fn shared_state() -> &'static (TestExecutionContext, GethNode) {
static STATE: LazyLock<(TestExecutionContext, GethNode)> = LazyLock::new(new_node);
&STATE
}
fn shared_node() -> &'static GethNode {
static NODE: LazyLock<(TestExecutionContext, GethNode)> = LazyLock::new(new_node);
&NODE.1
&shared_state().1
}
#[tokio::test]
async fn node_mines_simple_transfer_transaction_and_returns_receipt() {
// Arrange
let (context, node) = shared_state();
let account_address = context
.wallet_configuration
.wallet()
.default_signer()
.address();
let transaction = TransactionRequest::default()
.to(account_address)
.value(U256::from(100_000_000_000_000u128));
// Act
let receipt = node.execute_transaction(transaction).await;
// Assert
let _ = receipt.expect("Failed to get the receipt for the transfer");
}
#[test]
#[ignore = "Ignored since they take a long time to run"]
fn version_works() {
// Arrange
let node = shared_node();
@@ -670,6 +782,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_chain_id_from_node() {
// Arrange
let node = shared_node();
@@ -683,6 +796,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_gas_limit_from_node() {
// Arrange
let node = shared_node();
@@ -700,6 +814,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_coinbase_from_node() {
// Arrange
let node = shared_node();
@@ -717,6 +832,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_block_difficulty_from_node() {
// Arrange
let node = shared_node();
@@ -734,6 +850,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_block_hash_from_node() {
// Arrange
let node = shared_node();
@@ -751,6 +868,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_block_timestamp_from_node() {
// Arrange
let node = shared_node();
@@ -768,6 +886,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_block_number_from_node() {
// Arrange
let node = shared_node();
crates/node/src/lighthouse_geth.rs → crates/node/src/node_implementations/lighthouse_geth.rs
+366 -107
View File
@@ -9,7 +9,7 @@
//! that the tool has.
use std::{
collections::BTreeMap,
collections::{BTreeMap, HashSet},
fs::{File, create_dir_all},
io::Read,
ops::ControlFlow,
@@ -17,7 +17,7 @@ use std::{
pin::Pin,
process::{Command, Stdio},
sync::{
Arc,
Arc, LazyLock,
atomic::{AtomicU32, Ordering},
},
time::{Duration, SystemTime, UNIX_EPOCH},
@@ -31,20 +31,24 @@ use alloy::{
Address, BlockHash, BlockNumber, BlockTimestamp, StorageKey, TxHash, U256, address,
},
providers::{
Provider, ProviderBuilder,
Provider,
ext::DebugApi,
fillers::{CachedNonceManager, ChainIdFiller, FillProvider, NonceFiller, TxFiller},
},
rpc::types::{
EIP1186AccountProofResponse, TransactionRequest,
trace::geth::{DiffMode, GethDebugTracingOptions, PreStateConfig, PreStateFrame},
EIP1186AccountProofResponse, TransactionReceipt, TransactionRequest,
trace::geth::{
DiffMode, GethDebugTracingOptions, GethTrace, PreStateConfig, PreStateFrame,
},
},
};
use anyhow::Context as _;
use futures::{Stream, StreamExt};
use revive_common::EVMVersion;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use serde_with::serde_as;
use tracing::{Instrument, instrument};
use tokio::sync::{OnceCell, Semaphore};
use tracing::{Instrument, info, instrument};
use revive_dt_common::{
fs::clear_directory,
@@ -52,13 +56,13 @@ use revive_dt_common::{
};
use revive_dt_config::*;
use revive_dt_format::traits::ResolverApi;
use revive_dt_node_interaction::EthereumNode;
use revive_dt_node_interaction::{EthereumNode, MinedBlockInformation};
use crate::{
Node,
common::FallbackGasFiller,
constants::INITIAL_BALANCE,
process::{Process, ProcessReadinessWaitBehavior},
constants::{CHAIN_ID, INITIAL_BALANCE},
helpers::{Process, ProcessReadinessWaitBehavior},
provider_utils::{ConcreteProvider, FallbackGasFiller, construct_concurrency_limited_provider},
};
static NODE_COUNT: AtomicU32 = AtomicU32::new(0);
@@ -75,7 +79,8 @@ static NODE_COUNT: AtomicU32 = AtomicU32::new(0);
pub struct LighthouseGethNode {
/* Node Identifier */
id: u32,
connection_string: String,
ws_connection_string: String,
http_connection_string: String,
enclave_name: String,
/* Directory Paths */
@@ -91,17 +96,22 @@ pub struct LighthouseGethNode {
/* Spawned Processes */
process: Option<Process>,
/* Prefunded Account Information */
prefunded_account_address: Address,
/* Provider Related Fields */
wallet: Arc<EthereumWallet>,
nonce_manager: CachedNonceManager,
chain_id_filler: ChainIdFiller,
persistent_http_provider: OnceCell<ConcreteProvider<Ethereum, Arc<EthereumWallet>>>,
persistent_ws_provider: OnceCell<ConcreteProvider<Ethereum, Arc<EthereumWallet>>>,
http_provider_requests_semaphore: LazyLock<Semaphore>,
}
impl LighthouseGethNode {
const BASE_DIRECTORY: &str = "lighthouse";
const LOGS_DIRECTORY: &str = "logs";
const IPC_FILE_NAME: &str = "geth.ipc";
const CONFIG_FILE_NAME: &str = "config.yaml";
const TRANSACTION_INDEXING_ERROR: &str = "transaction indexing is in progress";
@@ -134,10 +144,8 @@ impl LighthouseGethNode {
Self {
/* Node Identifier */
id,
connection_string: base_directory
.join(Self::IPC_FILE_NAME)
.display()
.to_string(),
ws_connection_string: String::default(),
http_connection_string: String::default(),
enclave_name: format!(
"enclave-{}-{}",
SystemTime::now()
@@ -160,15 +168,20 @@ impl LighthouseGethNode {
/* Spawned Processes */
process: None,
/* Prefunded Account Information */
prefunded_account_address: wallet.default_signer().address(),
/* Provider Related Fields */
wallet: wallet.clone(),
nonce_manager: Default::default(),
chain_id_filler: Default::default(),
persistent_http_provider: OnceCell::const_new(),
persistent_ws_provider: OnceCell::const_new(),
http_provider_requests_semaphore: LazyLock::new(|| Semaphore::const_new(500)),
}
}
/// Create the node directory and call `geth init` to configure the genesis.
#[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
#[instrument(level = "info", skip_all, fields(lighthouse_node_id = self.id))]
fn init(&mut self, _: Genesis) -> anyhow::Result<&mut Self> {
self.init_directories()
.context("Failed to initialize the directories of the Lighthouse Geth node.")?;
@@ -198,10 +211,12 @@ impl LighthouseGethNode {
execution_layer_extra_parameters: vec![
"--nodiscover".to_string(),
"--cache=4096".to_string(),
"--txpool.globalslots=100000".to_string(),
"--txpool.globalqueue=100000".to_string(),
"--txpool.accountslots=128".to_string(),
"--txpool.accountqueue=1024".to_string(),
"--txlookuplimit=0".to_string(),
"--gcmode=archive".to_string(),
"--txpool.globalslots=500000".to_string(),
"--txpool.globalqueue=500000".to_string(),
"--txpool.accountslots=32768".to_string(),
"--txpool.accountqueue=32768".to_string(),
"--http.api=admin,engine,net,eth,web3,debug,txpool".to_string(),
"--http.addr=0.0.0.0".to_string(),
"--ws".to_string(),
@@ -211,13 +226,14 @@ impl LighthouseGethNode {
"--ws.origins=*".to_string(),
],
consensus_layer_extra_parameters: vec![
"--disable-quic".to_string(),
"--disable-deposit-contract-sync".to_string(),
],
}],
network_parameters: NetworkParameters {
preset: NetworkPreset::Mainnet,
seconds_per_slot: 12,
network_id: 420420420,
network_id: CHAIN_ID,
deposit_contract_address: address!("0x00000000219ab540356cBB839Cbe05303d7705Fa"),
altair_fork_epoch: 0,
bellatrix_fork_epoch: 0,
@@ -228,14 +244,8 @@ impl LighthouseGethNode {
num_validator_keys_per_node: 64,
genesis_delay: 10,
prefunded_accounts: {
let map = NetworkWallet::<Ethereum>::signer_addresses(&self.wallet)
.map(|address| {
(
address,
GenesisAccount::default()
.with_balance(INITIAL_BALANCE.try_into().unwrap()),
)
})
let map = std::iter::once(self.prefunded_account_address)
.map(|address| (address, GenesisAccount::default().with_balance(U256::MAX)))
.collect::<BTreeMap<_, _>>();
serde_json::to_string(&map).unwrap()
},
@@ -248,7 +258,12 @@ impl LighthouseGethNode {
public_port_start: Some(32000 + self.id as u16 * 1000),
},
),
consensus_layer_port_publisher_parameters: Default::default(),
consensus_layer_port_publisher_parameters: Some(
PortPublisherSingleItemParameters {
enabled: Some(true),
public_port_start: Some(59010 + self.id as u16 * 50),
},
),
}),
};
@@ -261,7 +276,7 @@ impl LighthouseGethNode {
}
/// Spawn the go-ethereum node child process.
#[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
#[instrument(level = "info", skip_all, fields(lighthouse_node_id = self.id))]
fn spawn_process(&mut self) -> anyhow::Result<&mut Self> {
let process = Process::new(
None,
@@ -292,6 +307,7 @@ impl LighthouseGethNode {
}),
},
)
.context("Failed to spawn the kurtosis enclave")
.inspect_err(|err| {
tracing::error!(?err, "Failed to spawn Kurtosis");
self.shutdown().expect("Failed to shutdown kurtosis");
@@ -316,64 +332,157 @@ impl LighthouseGethNode {
stdout
};
self.connection_string = stdout
self.http_connection_string = stdout
.split("el-1-geth-lighthouse")
.nth(1)
.and_then(|str| str.split(" rpc").nth(1))
.and_then(|str| str.split("->").nth(1))
.and_then(|str| str.split("\n").next())
.and_then(|str| str.trim().split(" ").next())
.map(|str| format!("http://{}", str.trim()))
.context("Failed to find the HTTP connection string of Kurtosis")?;
self.ws_connection_string = stdout
.split("el-1-geth-lighthouse")
.nth(1)
.and_then(|str| str.split("ws").nth(1))
.and_then(|str| str.split("->").nth(1))
.and_then(|str| str.split("\n").next())
.and_then(|str| str.trim().split(" ").next())
.map(|str| format!("ws://{}", str.trim()))
.context("Failed to find the WS connection string of Kurtosis")?;
info!(
http_connection_string = self.http_connection_string,
ws_connection_string = self.ws_connection_string,
"Discovered the connection strings for the node"
);
Ok(self)
}
async fn provider(
&self,
) -> anyhow::Result<FillProvider<impl TxFiller<Ethereum>, impl Provider<Ethereum>, Ethereum>>
{
ProviderBuilder::new()
.disable_recommended_fillers()
.filler(FallbackGasFiller::new(
25_000_000,
1_000_000_000,
1_000_000_000,
))
.filler(self.chain_id_filler.clone())
.filler(NonceFiller::new(self.nonce_manager.clone()))
.wallet(self.wallet.clone())
.connect(&self.connection_string)
.await
.context("Failed to create the provider for Kurtosis")
}
}
impl EthereumNode for LighthouseGethNode {
fn id(&self) -> usize {
self.id as _
}
fn connection_string(&self) -> &str {
&self.connection_string
}
#[instrument(
level = "info",
skip_all,
fields(geth_node_id = self.id, connection_string = self.connection_string),
err,
fields(lighthouse_node_id = self.id, connection_string = self.ws_connection_string),
err(Debug),
)]
fn execute_transaction(
&self,
transaction: TransactionRequest,
) -> Pin<Box<dyn Future<Output = anyhow::Result<alloy::rpc::types::TransactionReceipt>> + '_>>
{
Box::pin(async move {
let provider = self
.provider()
#[allow(clippy::type_complexity)]
async fn ws_provider(&self) -> anyhow::Result<ConcreteProvider<Ethereum, Arc<EthereumWallet>>> {
self.persistent_ws_provider
.get_or_try_init(|| async move {
construct_concurrency_limited_provider::<Ethereum, _>(
self.ws_connection_string.as_str(),
FallbackGasFiller::default(),
ChainIdFiller::new(Some(CHAIN_ID)),
NonceFiller::new(self.nonce_manager.clone()),
self.wallet.clone(),
)
.await
.context("Failed to create provider for transaction submission")?;
.context("Failed to construct the provider")
})
.await
.cloned()
}
#[instrument(
level = "info",
skip_all,
fields(lighthouse_node_id = self.id, connection_string = self.ws_connection_string),
err(Debug),
)]
#[allow(clippy::type_complexity)]
async fn http_provider(
&self,
) -> anyhow::Result<ConcreteProvider<Ethereum, Arc<EthereumWallet>>> {
self.persistent_http_provider
.get_or_try_init(|| async move {
construct_concurrency_limited_provider::<Ethereum, _>(
self.http_connection_string.as_str(),
FallbackGasFiller::default(),
ChainIdFiller::new(Some(CHAIN_ID)),
NonceFiller::new(self.nonce_manager.clone()),
self.wallet.clone(),
)
.await
.context("Failed to construct the provider")
})
.await
.cloned()
}
/// Funds all of the accounts in the Ethereum wallet from the initially funded account.
#[instrument(
level = "info",
skip_all,
fields(lighthouse_node_id = self.id, connection_string = self.ws_connection_string),
err(Debug),
)]
async fn fund_all_accounts(&self) -> anyhow::Result<()> {
let mut full_block_subscriber = self
.ws_provider()
.await
.context("Failed to create the WS provider")?
.subscribe_full_blocks()
.into_stream()
.await
.context("Full block subscriber")?;
let mut tx_hashes = futures::future::try_join_all(
NetworkWallet::<Ethereum>::signer_addresses(self.wallet.as_ref())
.enumerate()
.map(|(nonce, address)| async move {
let mut transaction = TransactionRequest::default()
.from(self.prefunded_account_address)
.to(address)
.nonce(nonce as _)
.value(INITIAL_BALANCE.try_into().unwrap());
transaction.chain_id = Some(CHAIN_ID);
self.submit_transaction(transaction).await
}),
)
.await
.context("Failed to submit all transactions")?
.into_iter()
.collect::<HashSet<_>>();
while let Some(block) = full_block_subscriber.next().await {
let Ok(block) = block else {
continue;
};
let block_number = block.number();
let block_timestamp = block.header.timestamp;
let block_transaction_count = block.transactions.len();
for hash in block.transactions.into_hashes().as_hashes().unwrap() {
tx_hashes.remove(hash);
}
info!(
block.number = block_number,
block.timestamp = block_timestamp,
block.transaction_count = block_transaction_count,
remaining_transactions = tx_hashes.len(),
"Discovered new block when funding accounts"
);
if tx_hashes.is_empty() {
break;
}
}
Ok(())
}
fn internal_execute_transaction<'a>(
transaction: TransactionRequest,
provider: FillProvider<
impl TxFiller<Ethereum> + 'a,
impl Provider<Ethereum> + Clone + 'a,
Ethereum,
>,
) -> Pin<Box<dyn Future<Output = anyhow::Result<TransactionReceipt>> + 'a>> {
Box::pin(async move {
let pending_transaction = provider
.send_transaction(transaction)
.await
@@ -404,10 +513,9 @@ impl EthereumNode for LighthouseGethNode {
// allow for a larger wait time. Therefore, in here we allow for 5 minutes of waiting
// with exponential backoff each time we attempt to get the receipt and find that it's
// not available.
let provider = Arc::new(provider);
poll(
Self::RECEIPT_POLLING_DURATION,
PollingWaitBehavior::Constant(Duration::from_millis(200)),
PollingWaitBehavior::Constant(Duration::from_millis(500)),
move || {
let provider = provider.clone();
async move {
@@ -432,17 +540,94 @@ impl EthereumNode for LighthouseGethNode {
.await
})
}
}
#[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
impl EthereumNode for LighthouseGethNode {
fn pre_transactions(&mut self) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + '_>> {
Box::pin(async move { self.fund_all_accounts().await })
}
fn id(&self) -> usize {
self.id as _
}
fn connection_string(&self) -> &str {
&self.ws_connection_string
}
#[instrument(
level = "info",
skip_all,
fields(lighthouse_node_id = self.id, connection_string = self.ws_connection_string),
err,
)]
fn submit_transaction(
&self,
transaction: TransactionRequest,
) -> Pin<Box<dyn Future<Output = anyhow::Result<TxHash>> + '_>> {
Box::pin(async move {
let _permit = self.http_provider_requests_semaphore.acquire().await;
let provider = self
.http_provider()
.await
.context("Failed to create the provider for transaction submission")?;
let pending_transaction = provider
.send_transaction(transaction)
.await
.context("Failed to submit the transaction through the provider")?;
Ok(*pending_transaction.tx_hash())
})
}
#[instrument(
level = "info",
skip_all,
fields(lighthouse_node_id = self.id, connection_string = self.ws_connection_string),
)]
fn get_receipt(
&self,
tx_hash: TxHash,
) -> Pin<Box<dyn Future<Output = anyhow::Result<TransactionReceipt>> + '_>> {
Box::pin(async move {
self.ws_provider()
.await
.context("Failed to create provider for getting the receipt")?
.get_transaction_receipt(tx_hash)
.await
.context("Failed to get the receipt of the transaction")?
.context("Failed to get the receipt of the transaction")
})
}
#[instrument(
level = "info",
skip_all,
fields(lighthouse_node_id = self.id, connection_string = self.ws_connection_string),
err,
)]
fn execute_transaction(
&self,
transaction: TransactionRequest,
) -> Pin<Box<dyn Future<Output = anyhow::Result<TransactionReceipt>> + '_>> {
Box::pin(async move {
let provider = self
.http_provider()
.await
.context("Failed to create provider for transaction execution")?;
Self::internal_execute_transaction(transaction, provider).await
})
}
#[instrument(level = "info", skip_all, fields(lighthouse_node_id = self.id))]
fn trace_transaction(
&self,
tx_hash: TxHash,
trace_options: GethDebugTracingOptions,
) -> Pin<Box<dyn Future<Output = anyhow::Result<alloy::rpc::types::trace::geth::GethTrace>> + '_>>
{
) -> Pin<Box<dyn Future<Output = anyhow::Result<GethTrace>> + '_>> {
Box::pin(async move {
let provider = Arc::new(
self.provider()
self.http_provider()
.await
.context("Failed to create provider for tracing")?,
);
@@ -473,7 +658,7 @@ impl EthereumNode for LighthouseGethNode {
})
}
#[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
#[instrument(level = "info", skip_all, fields(lighthouse_node_id = self.id))]
fn state_diff(
&self,
tx_hash: TxHash,
@@ -497,13 +682,13 @@ impl EthereumNode for LighthouseGethNode {
})
}
#[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
#[instrument(level = "info", skip_all, fields(lighthouse_node_id = self.id))]
fn balance_of(
&self,
address: Address,
) -> Pin<Box<dyn Future<Output = anyhow::Result<U256>> + '_>> {
Box::pin(async move {
self.provider()
self.ws_provider()
.await
.context("Failed to get the Geth provider")?
.get_balance(address)
@@ -512,14 +697,14 @@ impl EthereumNode for LighthouseGethNode {
})
}
#[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
#[instrument(level = "info", skip_all, fields(lighthouse_node_id = self.id))]
fn latest_state_proof(
&self,
address: Address,
keys: Vec<StorageKey>,
) -> Pin<Box<dyn Future<Output = anyhow::Result<EIP1186AccountProofResponse>> + '_>> {
Box::pin(async move {
self.provider()
self.ws_provider()
.await
.context("Failed to get the Geth provider")?
.get_proof(address, keys)
@@ -529,13 +714,13 @@ impl EthereumNode for LighthouseGethNode {
})
}
// #[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
#[instrument(level = "info", skip_all, fields(lighthouse_node_id = self.id))]
fn resolver(
&self,
) -> Pin<Box<dyn Future<Output = anyhow::Result<Arc<dyn ResolverApi + '_>>> + '_>> {
Box::pin(async move {
let id = self.id;
let provider = self.provider().await?;
let provider = self.ws_provider().await?;
Ok(Arc::new(LighthouseGethNodeResolver { id, provider }) as Arc<dyn ResolverApi>)
})
}
@@ -543,6 +728,43 @@ impl EthereumNode for LighthouseGethNode {
fn evm_version(&self) -> EVMVersion {
EVMVersion::Cancun
}
fn subscribe_to_full_blocks_information(
&self,
) -> Pin<
Box<
dyn Future<Output = anyhow::Result<Pin<Box<dyn Stream<Item = MinedBlockInformation>>>>>
+ '_,
>,
> {
Box::pin(async move {
let provider = self.ws_provider().await?;
let block_subscription = provider.subscribe_full_blocks().channel_size(1024);
let block_stream = block_subscription
.into_stream()
.await
.context("Failed to create the block stream")?;
let mined_block_information_stream = block_stream.filter_map(|block| async {
let block = block.ok()?;
Some(MinedBlockInformation {
block_number: block.number(),
block_timestamp: block.header.timestamp,
mined_gas: block.header.gas_used as _,
block_gas_limit: block.header.gas_limit as _,
transaction_hashes: block
.transactions
.into_hashes()
.as_hashes()
.expect("Must be hashes")
.to_vec(),
})
});
Ok(Box::pin(mined_block_information_stream)
as Pin<Box<dyn Stream<Item = MinedBlockInformation>>>)
})
}
}
pub struct LighthouseGethNodeResolver<F: TxFiller<Ethereum>, P: Provider<Ethereum>> {
@@ -553,14 +775,14 @@ pub struct LighthouseGethNodeResolver<F: TxFiller<Ethereum>, P: Provider<Ethereu
impl<F: TxFiller<Ethereum>, P: Provider<Ethereum>> ResolverApi
for LighthouseGethNodeResolver<F, P>
{
#[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
#[instrument(level = "info", skip_all, fields(lighthouse_node_id = self.id))]
fn chain_id(
&self,
) -> Pin<Box<dyn Future<Output = anyhow::Result<alloy::primitives::ChainId>> + '_>> {
Box::pin(async move { self.provider.get_chain_id().await.map_err(Into::into) })
}
#[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
#[instrument(level = "info", skip_all, fields(lighthouse_node_id = self.id))]
fn transaction_gas_price(
&self,
tx_hash: TxHash,
@@ -574,7 +796,7 @@ impl<F: TxFiller<Ethereum>, P: Provider<Ethereum>> ResolverApi
})
}
#[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
#[instrument(level = "info", skip_all, fields(lighthouse_node_id = self.id))]
fn block_gas_limit(
&self,
number: BlockNumberOrTag,
@@ -589,7 +811,7 @@ impl<F: TxFiller<Ethereum>, P: Provider<Ethereum>> ResolverApi
})
}
#[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
#[instrument(level = "info", skip_all, fields(lighthouse_node_id = self.id))]
fn block_coinbase(
&self,
number: BlockNumberOrTag,
@@ -604,7 +826,7 @@ impl<F: TxFiller<Ethereum>, P: Provider<Ethereum>> ResolverApi
})
}
#[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
#[instrument(level = "info", skip_all, fields(lighthouse_node_id = self.id))]
fn block_difficulty(
&self,
number: BlockNumberOrTag,
@@ -619,7 +841,7 @@ impl<F: TxFiller<Ethereum>, P: Provider<Ethereum>> ResolverApi
})
}
#[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
#[instrument(level = "info", skip_all, fields(lighthouse_node_id = self.id))]
fn block_base_fee(
&self,
number: BlockNumberOrTag,
@@ -639,7 +861,7 @@ impl<F: TxFiller<Ethereum>, P: Provider<Ethereum>> ResolverApi
})
}
#[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
#[instrument(level = "info", skip_all, fields(lighthouse_node_id = self.id))]
fn block_hash(
&self,
number: BlockNumberOrTag,
@@ -654,7 +876,7 @@ impl<F: TxFiller<Ethereum>, P: Provider<Ethereum>> ResolverApi
})
}
#[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
#[instrument(level = "info", skip_all, fields(lighthouse_node_id = self.id))]
fn block_timestamp(
&self,
number: BlockNumberOrTag,
@@ -669,29 +891,55 @@ impl<F: TxFiller<Ethereum>, P: Provider<Ethereum>> ResolverApi
})
}
#[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
#[instrument(level = "info", skip_all, fields(lighthouse_node_id = self.id))]
fn last_block_number(&self) -> Pin<Box<dyn Future<Output = anyhow::Result<BlockNumber>> + '_>> {
Box::pin(async move { self.provider.get_block_number().await.map_err(Into::into) })
}
}
impl Node for LighthouseGethNode {
#[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
#[instrument(level = "info", skip_all, fields(lighthouse_node_id = self.id))]
fn shutdown(&mut self) -> anyhow::Result<()> {
if !Command::new(self.kurtosis_binary_path.as_path())
let mut child = Command::new(self.kurtosis_binary_path.as_path())
.arg("enclave")
.arg("rm")
.arg("-f")
.arg(self.enclave_name.as_str())
.stdout(Stdio::null())
.stderr(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("Failed to spawn the enclave kill command")
.expect("Failed to spawn the enclave kill command");
if !child
.wait()
.expect("Failed to wait for the enclave kill command")
.success()
{
panic!("Failed to shut down the enclave {}", self.enclave_name)
let stdout = {
let mut stdout = String::default();
child
.stdout
.take()
.expect("Should be piped")
.read_to_string(&mut stdout)
.context("Failed to read stdout of kurtosis inspect to string")?;
stdout
};
let stderr = {
let mut stderr = String::default();
child
.stderr
.take()
.expect("Should be piped")
.read_to_string(&mut stderr)
.context("Failed to read stderr of kurtosis inspect to string")?;
stderr
};
panic!(
"Failed to shut down the enclave {} - stdout: {stdout}, stderr: {stderr}",
self.enclave_name
)
}
drop(self.process.take());
@@ -699,13 +947,13 @@ impl Node for LighthouseGethNode {
Ok(())
}
#[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
#[instrument(level = "info", skip_all, fields(lighthouse_node_id = self.id))]
fn spawn(&mut self, genesis: Genesis) -> anyhow::Result<()> {
self.init(genesis)?.spawn_process()?;
Ok(())
}
#[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
#[instrument(level = "info", skip_all, fields(lighthouse_node_id = self.id))]
fn version(&self) -> anyhow::Result<String> {
let output = Command::new(&self.kurtosis_binary_path)
.arg("version")
@@ -722,7 +970,7 @@ impl Node for LighthouseGethNode {
}
impl Drop for LighthouseGethNode {
#[instrument(level = "info", skip_all, fields(geth_node_id = self.id))]
#[instrument(level = "info", skip_all, fields(lighthouse_node_id = self.id))]
fn drop(&mut self) {
self.shutdown().expect("Failed to shutdown")
}
@@ -853,7 +1101,9 @@ mod tests {
use super::*;
fn test_config() -> TestExecutionContext {
TestExecutionContext::default()
let mut config = TestExecutionContext::default();
config.wallet_configuration.additional_keys = 100;
config
}
fn new_node() -> (TestExecutionContext, LighthouseGethNode) {
@@ -888,6 +1138,7 @@ mod tests {
async fn node_mines_simple_transfer_transaction_and_returns_receipt() {
// Arrange
let (context, node) = new_node();
node.fund_all_accounts().await.expect("Failed");
let account_address = context
.wallet_configuration
@@ -906,6 +1157,7 @@ mod tests {
}
#[test]
#[ignore = "Ignored since they take a long time to run"]
fn version_works() {
// Arrange
let (_context, node) = new_node();
@@ -922,6 +1174,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_chain_id_from_node() {
// Arrange
let (_context, node) = new_node();
@@ -935,6 +1188,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_gas_limit_from_node() {
// Arrange
let (_context, node) = new_node();
@@ -952,6 +1206,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_coinbase_from_node() {
// Arrange
let (_context, node) = new_node();
@@ -969,6 +1224,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_block_difficulty_from_node() {
// Arrange
let (_context, node) = new_node();
@@ -986,6 +1242,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_block_hash_from_node() {
// Arrange
let (_context, node) = new_node();
@@ -1003,6 +1260,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_block_timestamp_from_node() {
// Arrange
let (_context, node) = new_node();
@@ -1020,6 +1278,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_block_number_from_node() {
// Arrange
let (_context, node) = new_node();
crates/node/src/node_implementations/mod.rs
+3
View File
@@ -0,0 +1,3 @@
pub mod geth;
pub mod lighthouse_geth;
pub mod substrate;
crates/node/src/substrate.rs → crates/node/src/node_implementations/substrate.rs
+133 -38
View File
@@ -23,17 +23,20 @@ use alloy::{
TxHash, U256,
},
providers::{
Provider, ProviderBuilder,
Provider,
ext::DebugApi,
fillers::{CachedNonceManager, ChainIdFiller, FillProvider, NonceFiller, TxFiller},
fillers::{CachedNonceManager, ChainIdFiller, NonceFiller},
},
rpc::types::{
EIP1186AccountProofResponse, TransactionReceipt,
EIP1186AccountProofResponse, TransactionReceipt, TransactionRequest,
eth::{Block, Header, Transaction},
trace::geth::{DiffMode, GethDebugTracingOptions, PreStateConfig, PreStateFrame},
trace::geth::{
DiffMode, GethDebugTracingOptions, GethTrace, PreStateConfig, PreStateFrame,
},
},
};
use anyhow::Context as _;
use futures::{Stream, StreamExt};
use revive_common::EVMVersion;
use revive_dt_common::fs::clear_directory;
use revive_dt_format::traits::ResolverApi;
@@ -43,14 +46,15 @@ use sp_core::crypto::Ss58Codec;
use sp_runtime::AccountId32;
use revive_dt_config::*;
use revive_dt_node_interaction::EthereumNode;
use revive_dt_node_interaction::{EthereumNode, MinedBlockInformation};
use tokio::sync::OnceCell;
use tracing::instrument;
use crate::{
Node,
common::FallbackGasFiller,
constants::INITIAL_BALANCE,
process::{Process, ProcessReadinessWaitBehavior},
constants::{CHAIN_ID, INITIAL_BALANCE},
helpers::{Process, ProcessReadinessWaitBehavior},
provider_utils::{ConcreteProvider, FallbackGasFiller, construct_concurrency_limited_provider},
};
static NODE_COUNT: AtomicU32 = AtomicU32::new(0);
@@ -59,6 +63,7 @@ static NODE_COUNT: AtomicU32 = AtomicU32::new(0);
/// or the revive-dev-node which is done by changing the path and some of the other arguments passed
/// to the command.
#[derive(Debug)]
pub struct SubstrateNode {
id: u32,
node_binary: PathBuf,
@@ -71,7 +76,7 @@ pub struct SubstrateNode {
eth_proxy_process: Option<Process>,
wallet: Arc<EthereumWallet>,
nonce_manager: CachedNonceManager,
chain_id_filler: ChainIdFiller,
provider: OnceCell<ConcreteProvider<ReviveNetwork, Arc<EthereumWallet>>>,
}
impl SubstrateNode {
@@ -121,8 +126,8 @@ impl SubstrateNode {
substrate_process: None,
eth_proxy_process: None,
wallet: wallet.clone(),
chain_id_filler: Default::default(),
nonce_manager: Default::default(),
provider: Default::default(),
}
}
@@ -144,6 +149,7 @@ impl SubstrateNode {
.arg(self.export_chainspec_command.as_str())
.arg("--chain")
.arg("dev")
.env_remove("RUST_LOG")
.output()
.context("Failed to export the chain-spec")?;
@@ -335,27 +341,29 @@ impl SubstrateNode {
async fn provider(
&self,
) -> anyhow::Result<
FillProvider<impl TxFiller<ReviveNetwork>, impl Provider<ReviveNetwork>, ReviveNetwork>,
> {
ProviderBuilder::new()
.disable_recommended_fillers()
.network::<ReviveNetwork>()
.filler(FallbackGasFiller::new(
25_000_000,
1_000_000_000,
1_000_000_000,
))
.filler(self.chain_id_filler.clone())
.filler(NonceFiller::new(self.nonce_manager.clone()))
.wallet(self.wallet.clone())
.connect(&self.rpc_url)
) -> anyhow::Result<ConcreteProvider<ReviveNetwork, Arc<EthereumWallet>>> {
self.provider
.get_or_try_init(|| async move {
construct_concurrency_limited_provider::<ReviveNetwork, _>(
self.rpc_url.as_str(),
FallbackGasFiller::new(250_000_000, 5_000_000_000, 1_000_000_000),
ChainIdFiller::new(Some(CHAIN_ID)),
NonceFiller::new(self.nonce_manager.clone()),
self.wallet.clone(),
)
.await
.context("Failed to construct the provider")
})
.await
.map_err(Into::into)
.cloned()
}
}
impl EthereumNode for SubstrateNode {
fn pre_transactions(&mut self) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + '_>> {
Box::pin(async move { Ok(()) })
}
fn id(&self) -> usize {
self.id as _
}
@@ -364,11 +372,48 @@ impl EthereumNode for SubstrateNode {
&self.rpc_url
}
fn execute_transaction(
fn submit_transaction(
&self,
transaction: alloy::rpc::types::TransactionRequest,
transaction: TransactionRequest,
) -> Pin<Box<dyn Future<Output = anyhow::Result<TxHash>> + '_>> {
Box::pin(async move {
let provider = self
.provider()
.await
.context("Failed to create the provider for transaction submission")?;
let pending_transaction = provider
.send_transaction(transaction)
.await
.context("Failed to submit the transaction through the provider")?;
Ok(*pending_transaction.tx_hash())
})
}
fn get_receipt(
&self,
tx_hash: TxHash,
) -> Pin<Box<dyn Future<Output = anyhow::Result<TransactionReceipt>> + '_>> {
Box::pin(async move {
self.provider()
.await
.context("Failed to create provider for getting the receipt")?
.get_transaction_receipt(tx_hash)
.await
.context("Failed to get the receipt of the transaction")?
.context("Failed to get the receipt of the transaction")
})
}
fn execute_transaction(
&self,
transaction: TransactionRequest,
) -> Pin<Box<dyn Future<Output = anyhow::Result<TransactionReceipt>> + '_>> {
static SEMAPHORE: std::sync::LazyLock<tokio::sync::Semaphore> =
std::sync::LazyLock::new(|| tokio::sync::Semaphore::new(500));
Box::pin(async move {
let _permit = SEMAPHORE.acquire().await?;
let receipt = self
.provider()
.await
@@ -387,8 +432,7 @@ impl EthereumNode for SubstrateNode {
&self,
tx_hash: TxHash,
trace_options: GethDebugTracingOptions,
) -> Pin<Box<dyn Future<Output = anyhow::Result<alloy::rpc::types::trace::geth::GethTrace>> + '_>>
{
) -> Pin<Box<dyn Future<Output = anyhow::Result<GethTrace>> + '_>> {
Box::pin(async move {
self.provider()
.await
@@ -463,16 +507,56 @@ impl EthereumNode for SubstrateNode {
fn evm_version(&self) -> EVMVersion {
EVMVersion::Cancun
}
fn subscribe_to_full_blocks_information(
&self,
) -> Pin<
Box<
dyn Future<Output = anyhow::Result<Pin<Box<dyn Stream<Item = MinedBlockInformation>>>>>
+ '_,
>,
> {
Box::pin(async move {
let provider = self
.provider()
.await
.context("Failed to create the provider for block subscription")?;
let mut block_subscription = provider
.watch_full_blocks()
.await
.context("Failed to create the blocks stream")?;
block_subscription.set_channel_size(0xFFFF);
block_subscription.set_poll_interval(Duration::from_secs(1));
let block_stream = block_subscription.into_stream();
let mined_block_information_stream = block_stream.filter_map(|block| async {
let block = block.ok()?;
Some(MinedBlockInformation {
block_number: block.number(),
block_timestamp: block.header.timestamp,
mined_gas: block.header.gas_used as _,
block_gas_limit: block.header.gas_limit,
transaction_hashes: block
.transactions
.into_hashes()
.as_hashes()
.expect("Must be hashes")
.to_vec(),
})
});
Ok(Box::pin(mined_block_information_stream)
as Pin<Box<dyn Stream<Item = MinedBlockInformation>>>)
})
}
}
pub struct SubstrateNodeResolver<F: TxFiller<ReviveNetwork>, P: Provider<ReviveNetwork>> {
pub struct SubstrateNodeResolver {
id: u32,
provider: FillProvider<F, P, ReviveNetwork>,
provider: ConcreteProvider<ReviveNetwork, Arc<EthereumWallet>>,
}
impl<F: TxFiller<ReviveNetwork>, P: Provider<ReviveNetwork>> ResolverApi
for SubstrateNodeResolver<F, P>
{
impl ResolverApi for SubstrateNodeResolver {
#[instrument(level = "info", skip_all, fields(substrate_node_id = self.id))]
fn chain_id(
&self,
@@ -1068,9 +1152,7 @@ mod tests {
use crate::Node;
fn test_config() -> TestExecutionContext {
let mut context = TestExecutionContext::default();
context.kitchensink_configuration.use_kitchensink = true;
context
TestExecutionContext::default()
}
fn new_node() -> (TestExecutionContext, SubstrateNode) {
@@ -1142,6 +1224,7 @@ mod tests {
}
#[test]
#[ignore = "Ignored since they take a long time to run"]
fn test_init_generates_chainspec_with_balances() {
let genesis_content = r#"
{
@@ -1195,6 +1278,7 @@ mod tests {
}
#[test]
#[ignore = "Ignored since they take a long time to run"]
fn test_parse_genesis_alloc() {
// Create test genesis file
let genesis_json = r#"
@@ -1237,6 +1321,7 @@ mod tests {
}
#[test]
#[ignore = "Ignored since they take a long time to run"]
fn print_eth_to_substrate_mappings() {
let eth_addresses = vec![
"0x90F8bf6A479f320ead074411a4B0e7944Ea8c9C1",
@@ -1252,6 +1337,7 @@ mod tests {
}
#[test]
#[ignore = "Ignored since they take a long time to run"]
fn test_eth_to_substrate_address() {
let cases = vec![
(
@@ -1282,6 +1368,7 @@ mod tests {
}
#[test]
#[ignore = "Ignored since they take a long time to run"]
fn version_works() {
let node = shared_node();
@@ -1294,6 +1381,7 @@ mod tests {
}
#[test]
#[ignore = "Ignored since they take a long time to run"]
fn eth_rpc_version_works() {
let node = shared_node();
@@ -1306,6 +1394,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_chain_id_from_node() {
// Arrange
let node = shared_node();
@@ -1319,6 +1408,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_gas_limit_from_node() {
// Arrange
let node = shared_node();
@@ -1336,6 +1426,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_coinbase_from_node() {
// Arrange
let node = shared_node();
@@ -1353,6 +1444,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_block_difficulty_from_node() {
// Arrange
let node = shared_node();
@@ -1370,6 +1462,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_block_hash_from_node() {
// Arrange
let node = shared_node();
@@ -1387,6 +1480,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_block_timestamp_from_node() {
// Arrange
let node = shared_node();
@@ -1404,6 +1498,7 @@ mod tests {
}
#[tokio::test]
#[ignore = "Ignored since they take a long time to run"]
async fn can_get_block_number_from_node() {
// Arrange
let node = shared_node();
crates/node/src/provider_utils/concurrency_limiter.rs
+69
View File
@@ -0,0 +1,69 @@
use std::sync::Arc;
use alloy::transports::BoxFuture;
use tokio::sync::Semaphore;
use tower::{Layer, Service};
#[derive(Clone, Debug)]
pub struct ConcurrencyLimiterLayer {
semaphore: Arc<Semaphore>,
}
impl ConcurrencyLimiterLayer {
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>;
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>,
}
impl<S, Request> Service<Request> for ConcurrencyLimiterService<S>
where
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>>;
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);
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
})
}
}
crates/node/src/common.rs → crates/node/src/provider_utils/fallback_gas_provider.rs
+6
View File
@@ -30,6 +30,12 @@ impl FallbackGasFiller {
}
}
impl Default for FallbackGasFiller {
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,
crates/node/src/provider_utils/mod.rs
+7
View File
@@ -0,0 +1,7 @@
mod concurrency_limiter;
mod fallback_gas_provider;
mod provider;
pub use concurrency_limiter::*;
pub use fallback_gas_provider::*;
pub use provider::*;
crates/node/src/provider_utils/provider.rs
+63
View File
@@ -0,0 +1,63 @@
use std::sync::LazyLock;
use alloy::{
network::{Network, NetworkWallet, TransactionBuilder4844},
providers::{
Identity, ProviderBuilder, RootProvider,
fillers::{ChainIdFiller, FillProvider, JoinFill, NonceFiller, TxFiller, WalletFiller},
},
rpc::client::ClientBuilder,
};
use anyhow::{Context, Result};
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,
>;
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,
) -> 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>,
{
// 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 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)
}