Fix the Fallback Gas Limiter (#217)

* Add code to disable the fallback gas filler

* Allow benchmarks driver to await tx receipts

* Improve the transaction submission logic

* Update Python Script to process Geth benchmarks
This commit is contained in:
Omar
2025-12-04 16:19:48 +03:00
committed by GitHub
parent 3edaebdcae
commit 78ac7ee381
11 changed files with 269 additions and 85 deletions
+17
View File
@@ -375,6 +375,23 @@ pub struct BenchmarkingContext {
#[arg(short = 'r', long = "default-repetition-count", default_value_t = 1000)]
pub default_repetition_count: usize,
/// This transaction controls whether the benchmarking driver should await for transactions to
/// be included in a block before moving on to the next transaction in the sequence or not.
///
/// This behavior is useful in certain cases and not so useful in others. For example, in some
/// repetition block if there's some kind of relationship between txs n and n+1 (for example a
/// mint then a transfer) then you would want to wait for the minting to happen and then move on
/// to the transfers. On the other hand, if there's no relationship between the transactions n
/// and n+1 (e.g., mint and another mint of a different token) then awaiting the first mint to
/// be included in a block might not seem necessary.
///
/// By default, this behavior is set to false to allow the benchmarking framework to saturate
/// the node's mempool as quickly as possible. However, as explained above, there are cases
/// where it's needed and certain workloads where failure to provide this argument would lead to
/// inaccurate results.
#[arg(long)]
pub await_transaction_inclusion: bool,
/// Configuration parameters for the corpus files to use.
#[clap(flatten, next_help_heading = "Corpus Configuration")]
pub corpus_configuration: CorpusConfiguration,
@@ -1,6 +1,5 @@
use std::{
collections::HashMap,
ops::ControlFlow,
sync::{
Arc,
atomic::{AtomicUsize, Ordering},
@@ -13,6 +12,7 @@ use alloy::{
json_abi::JsonAbi,
network::{Ethereum, TransactionBuilder},
primitives::{Address, TxHash, U256},
providers::Provider,
rpc::types::{
TransactionReceipt, TransactionRequest,
trace::geth::{
@@ -22,12 +22,9 @@ use alloy::{
},
};
use anyhow::{Context as _, Result, bail};
use futures::TryFutureExt;
use futures::{FutureExt as _, TryFutureExt};
use indexmap::IndexMap;
use revive_dt_common::{
futures::{PollingWaitBehavior, poll},
types::PrivateKeyAllocator,
};
use revive_dt_common::types::PrivateKeyAllocator;
use revive_dt_format::{
metadata::{ContractInstance, ContractPathAndIdent},
steps::{
@@ -37,7 +34,7 @@ use revive_dt_format::{
traits::{ResolutionContext, ResolverApi},
};
use tokio::sync::{Mutex, OnceCell, mpsc::UnboundedSender};
use tracing::{Instrument, Span, debug, error, field::display, info, info_span, instrument};
use tracing::{Span, debug, error, field::display, info, instrument};
use crate::{
differential_benchmarks::{ExecutionState, WatcherEvent},
@@ -73,6 +70,10 @@ pub struct Driver<'a, I> {
/// The number of steps that were executed on the driver.
steps_executed: usize,
/// This function controls if the driver should wait for transactions to be included in a block
/// or not before proceeding forward.
await_transaction_inclusion: bool,
/// This is the queue of steps that are to be executed by the driver for this test case. Each
/// time `execute_step` is called one of the steps is executed.
steps_iterator: I,
@@ -89,6 +90,7 @@ where
private_key_allocator: Arc<Mutex<PrivateKeyAllocator>>,
cached_compiler: &CachedCompiler<'a>,
watcher_tx: UnboundedSender<WatcherEvent>,
await_transaction_inclusion: bool,
steps: I,
) -> Result<Self> {
let mut this = Driver {
@@ -104,6 +106,7 @@ where
execution_state: ExecutionState::empty(),
steps_executed: 0,
steps_iterator: steps,
await_transaction_inclusion,
watcher_tx,
};
this.init_execution_state(cached_compiler)
@@ -166,7 +169,7 @@ where
code,
);
let receipt = self
.execute_transaction(tx, None)
.execute_transaction(tx, None, Duration::from_secs(5 * 60))
.and_then(|(_, receipt_fut)| receipt_fut)
.await
.inspect_err(|err| {
@@ -365,7 +368,30 @@ where
let tx = step
.as_transaction(self.resolver.as_ref(), self.default_resolution_context())
.await?;
Ok(self.execute_transaction(tx, Some(step_path)).await?.0)
let (tx_hash, receipt_future) = self
.execute_transaction(tx.clone(), Some(step_path), Duration::from_secs(30 * 60))
.await?;
if self.await_transaction_inclusion {
let receipt = receipt_future
.await
.context("Failed while waiting for transaction inclusion in block")?;
if !receipt.status() {
error!(
?tx,
tx.hash = %receipt.transaction_hash,
?receipt,
"Encountered a failing benchmark transaction"
);
bail!(
"Encountered a failing transaction in benchmarks: {}",
receipt.transaction_hash
)
}
}
Ok(tx_hash)
}
}
}
@@ -466,6 +492,7 @@ where
.collect::<Vec<_>>();
steps.into_iter()
},
await_transaction_inclusion: self.await_transaction_inclusion,
watcher_tx: self.watcher_tx.clone(),
})
.map(|driver| driver.execute_all());
@@ -632,7 +659,7 @@ where
};
let receipt = match self
.execute_transaction(tx, step_path)
.execute_transaction(tx, step_path, Duration::from_secs(5 * 60))
.and_then(|(_, receipt_fut)| receipt_fut)
.await
{
@@ -677,18 +704,33 @@ where
#[instrument(
level = "info",
skip_all,
fields(driver_id = self.driver_id, transaction_hash = tracing::field::Empty)
fields(
driver_id = self.driver_id,
transaction = ?transaction,
transaction_hash = tracing::field::Empty
),
err(Debug)
)]
async fn execute_transaction(
&self,
transaction: TransactionRequest,
step_path: Option<&StepPath>,
receipt_wait_duration: Duration,
) -> anyhow::Result<(TxHash, impl Future<Output = Result<TransactionReceipt>>)> {
let node = self.platform_information.node;
let transaction_hash = node
.submit_transaction(transaction)
let provider = node.provider().await.context("Creating provider failed")?;
let pending_transaction_builder = provider
.send_transaction(transaction)
.await
.context("Failed to submit transaction")?;
let transaction_hash = *pending_transaction_builder.tx_hash();
let receipt_future = pending_transaction_builder
.with_timeout(Some(receipt_wait_duration))
.with_required_confirmations(2)
.get_receipt()
.map(|res| res.context("Failed to get the receipt of the transaction"));
Span::current().record("transaction_hash", display(transaction_hash));
info!("Submitted transaction");
@@ -701,28 +743,7 @@ where
.context("Failed to send the transaction hash to the watcher")?;
};
Ok((transaction_hash, async move {
info!("Starting to poll for transaction receipt");
poll(
Duration::from_secs(30 * 60),
PollingWaitBehavior::Constant(Duration::from_secs(1)),
|| {
async move {
match node.get_receipt(transaction_hash).await {
Ok(receipt) => {
info!("Polling succeeded, receipt found");
Ok(ControlFlow::Break(receipt))
}
Err(_) => Ok(ControlFlow::Continue(())),
}
}
.instrument(info_span!("Polling for receipt"))
},
)
.instrument(info_span!("Polling for receipt", %transaction_hash))
.await
.inspect(|_| info!("Found the transaction receipt"))
}))
Ok((transaction_hash, receipt_future))
}
// endregion:Transaction Execution
}
@@ -160,6 +160,7 @@ pub async fn handle_differential_benchmarks(
private_key_allocator,
cached_compiler.as_ref(),
watcher_tx.clone(),
context.await_transaction_inclusion,
test_definition
.case
.steps_iterator_for_benchmarks(context.default_repetition_count)
@@ -139,23 +139,18 @@ impl Watcher {
break;
}
info!(
block_number = block.ethereum_block_information.block_number,
block_tx_count = block.ethereum_block_information.transaction_hashes.len(),
remaining_transactions = watch_for_transaction_hashes.read().await.len(),
"Observed a block"
);
// Remove all of the transaction hashes observed in this block from the txs we
// are currently watching for.
let mut watch_for_transaction_hashes =
watch_for_transaction_hashes.write().await;
let mut relevant_transactions_observed = 0;
for tx_hash in block.ethereum_block_information.transaction_hashes.iter() {
let Some((step_path, submission_time)) =
watch_for_transaction_hashes.remove(tx_hash)
else {
continue;
};
relevant_transactions_observed += 1;
let transaction_information = TransactionInformation {
transaction_hash: *tx_hash,
submission_timestamp: submission_time
@@ -172,6 +167,14 @@ impl Watcher {
)
.expect("Can't fail")
}
info!(
block_number = block.ethereum_block_information.block_number,
block_tx_count = block.ethereum_block_information.transaction_hashes.len(),
relevant_transactions_observed,
remaining_transactions = watch_for_transaction_hashes.len(),
"Observed a block"
);
}
info!("Watcher's Block Watching Task Finished");
+14 -4
View File
@@ -91,7 +91,8 @@ impl Platform for GethEvmSolcPlatform {
let genesis_configuration = AsRef::<GenesisConfiguration>::as_ref(&context);
let genesis = genesis_configuration.genesis()?.clone();
Ok(thread::spawn(move || {
let node = GethNode::new(context);
let use_fallback_gas_filler = matches!(context, Context::Test(..));
let node = GethNode::new(context, use_fallback_gas_filler);
let node = spawn_node::<GethNode>(node, genesis)?;
Ok(Box::new(node) as Box<_>)
}))
@@ -145,7 +146,8 @@ impl Platform for LighthouseGethEvmSolcPlatform {
let genesis_configuration = AsRef::<GenesisConfiguration>::as_ref(&context);
let genesis = genesis_configuration.genesis()?.clone();
Ok(thread::spawn(move || {
let node = LighthouseGethNode::new(context);
let use_fallback_gas_filler = matches!(context, Context::Test(..));
let node = LighthouseGethNode::new(context, use_fallback_gas_filler);
let node = spawn_node::<LighthouseGethNode>(node, genesis)?;
Ok(Box::new(node) as Box<_>)
}))
@@ -206,12 +208,14 @@ impl Platform for ReviveDevNodePolkavmResolcPlatform {
let genesis = genesis_configuration.genesis()?.clone();
Ok(thread::spawn(move || {
let use_fallback_gas_filler = matches!(context, Context::Test(..));
let node = SubstrateNode::new(
revive_dev_node_path,
SubstrateNode::REVIVE_DEV_NODE_EXPORT_CHAINSPEC_COMMAND,
Some(revive_dev_node_consensus),
context,
&eth_rpc_connection_strings,
use_fallback_gas_filler,
);
let node = spawn_node(node, genesis)?;
Ok(Box::new(node) as Box<_>)
@@ -274,12 +278,14 @@ impl Platform for ReviveDevNodeRevmSolcPlatform {
let genesis = genesis_configuration.genesis()?.clone();
Ok(thread::spawn(move || {
let use_fallback_gas_filler = matches!(context, Context::Test(..));
let node = SubstrateNode::new(
revive_dev_node_path,
SubstrateNode::REVIVE_DEV_NODE_EXPORT_CHAINSPEC_COMMAND,
Some(revive_dev_node_consensus),
context,
&eth_rpc_connection_strings,
use_fallback_gas_filler,
);
let node = spawn_node(node, genesis)?;
Ok(Box::new(node) as Box<_>)
@@ -338,7 +344,9 @@ impl Platform for ZombienetPolkavmResolcPlatform {
.clone();
let genesis = genesis_configuration.genesis()?.clone();
Ok(thread::spawn(move || {
let node = ZombienetNode::new(polkadot_parachain_path, context);
let use_fallback_gas_filler = matches!(context, Context::Test(..));
let node =
ZombienetNode::new(polkadot_parachain_path, context, use_fallback_gas_filler);
let node = spawn_node(node, genesis)?;
Ok(Box::new(node) as Box<_>)
}))
@@ -395,7 +403,9 @@ impl Platform for ZombienetRevmSolcPlatform {
.clone();
let genesis = genesis_configuration.genesis()?.clone();
Ok(thread::spawn(move || {
let node = ZombienetNode::new(polkadot_parachain_path, context);
let use_fallback_gas_filler = matches!(context, Context::Test(..));
let node =
ZombienetNode::new(polkadot_parachain_path, context, use_fallback_gas_filler);
let node = spawn_node(node, genesis)?;
Ok(Box::new(node) as Box<_>)
}))
+6 -2
View File
@@ -76,6 +76,7 @@ pub struct GethNode {
wallet: Arc<EthereumWallet>,
nonce_manager: CachedNonceManager,
provider: OnceCell<ConcreteProvider<Ethereum, Arc<EthereumWallet>>>,
use_fallback_gas_filler: bool,
}
impl GethNode {
@@ -100,6 +101,7 @@ impl GethNode {
+ AsRef<WalletConfiguration>
+ AsRef<GethConfiguration>
+ Clone,
use_fallback_gas_filler: bool,
) -> Self {
let working_directory_configuration =
AsRef::<WorkingDirectoryConfiguration>::as_ref(&context);
@@ -126,6 +128,7 @@ impl GethNode {
wallet: wallet.clone(),
nonce_manager: Default::default(),
provider: Default::default(),
use_fallback_gas_filler,
}
}
@@ -246,7 +249,8 @@ impl GethNode {
.get_or_try_init(|| async move {
construct_concurrency_limited_provider::<Ethereum, _>(
self.connection_string.as_str(),
FallbackGasFiller::default(),
FallbackGasFiller::default()
.with_use_fallback_gas_filler(self.use_fallback_gas_filler),
ChainIdFiller::new(Some(CHAIN_ID)),
NonceFiller::new(self.nonce_manager.clone()),
self.wallet.clone(),
@@ -742,7 +746,7 @@ mod tests {
fn new_node() -> (TestExecutionContext, GethNode) {
let context = test_config();
let mut node = GethNode::new(&context);
let mut node = GethNode::new(&context, true);
node.init(context.genesis_configuration.genesis().unwrap().clone())
.expect("Failed to initialize the node")
.spawn_process()
@@ -106,6 +106,8 @@ pub struct LighthouseGethNode {
persistent_http_provider: OnceCell<ConcreteProvider<Ethereum, Arc<EthereumWallet>>>,
persistent_ws_provider: OnceCell<ConcreteProvider<Ethereum, Arc<EthereumWallet>>>,
use_fallback_gas_filler: bool,
}
impl LighthouseGethNode {
@@ -127,6 +129,7 @@ impl LighthouseGethNode {
+ AsRef<WalletConfiguration>
+ AsRef<KurtosisConfiguration>
+ Clone,
use_fallback_gas_filler: bool,
) -> Self {
let working_directory_configuration =
AsRef::<WorkingDirectoryConfiguration>::as_ref(&context);
@@ -176,6 +179,7 @@ impl LighthouseGethNode {
nonce_manager: Default::default(),
persistent_http_provider: OnceCell::const_new(),
persistent_ws_provider: OnceCell::const_new(),
use_fallback_gas_filler,
}
}
@@ -374,7 +378,8 @@ impl LighthouseGethNode {
.get_or_try_init(|| async move {
construct_concurrency_limited_provider::<Ethereum, _>(
self.ws_connection_string.as_str(),
FallbackGasFiller::default(),
FallbackGasFiller::default()
.with_use_fallback_gas_filler(self.use_fallback_gas_filler),
ChainIdFiller::new(Some(CHAIN_ID)),
NonceFiller::new(self.nonce_manager.clone()),
self.wallet.clone(),
@@ -1152,7 +1157,7 @@ mod tests {
let _guard = NODE_START_MUTEX.lock().unwrap();
let context = test_config();
let mut node = LighthouseGethNode::new(&context);
let mut node = LighthouseGethNode::new(&context, true);
node.init(context.genesis_configuration.genesis().unwrap().clone())
.expect("Failed to initialize the node")
.spawn_process()
@@ -79,6 +79,7 @@ pub struct SubstrateNode {
nonce_manager: CachedNonceManager,
provider: OnceCell<ConcreteProvider<Ethereum, Arc<EthereumWallet>>>,
consensus: Option<String>,
use_fallback_gas_filler: bool,
}
impl SubstrateNode {
@@ -105,6 +106,7 @@ impl SubstrateNode {
+ AsRef<EthRpcConfiguration>
+ AsRef<WalletConfiguration>,
existing_connection_strings: &[String],
use_fallback_gas_filler: bool,
) -> Self {
let working_directory_path =
AsRef::<WorkingDirectoryConfiguration>::as_ref(&context).as_path();
@@ -137,6 +139,7 @@ impl SubstrateNode {
nonce_manager: Default::default(),
provider: Default::default(),
consensus,
use_fallback_gas_filler,
}
}
@@ -324,7 +327,12 @@ impl SubstrateNode {
.get_or_try_init(|| async move {
construct_concurrency_limited_provider::<Ethereum, _>(
self.rpc_url.as_str(),
FallbackGasFiller::new(u64::MAX, 50_000_000_000, 1_000_000_000),
FallbackGasFiller::new(
u64::MAX,
50_000_000_000,
1_000_000_000,
self.use_fallback_gas_filler,
),
ChainIdFiller::new(Some(CHAIN_ID)),
NonceFiller::new(self.nonce_manager.clone()),
self.wallet.clone(),
@@ -825,6 +833,7 @@ mod tests {
None,
&context,
&[],
true,
);
node.init(context.genesis_configuration.genesis().unwrap().clone())
.expect("Failed to initialize the node")
@@ -896,6 +905,7 @@ mod tests {
None,
&context,
&[],
true,
);
// Call `init()`
@@ -114,6 +114,8 @@ pub struct ZombienetNode {
nonce_manager: CachedNonceManager,
provider: OnceCell<ConcreteProvider<Ethereum, Arc<EthereumWallet>>>,
use_fallback_gas_filler: bool,
}
impl ZombienetNode {
@@ -137,6 +139,7 @@ impl ZombienetNode {
context: impl AsRef<WorkingDirectoryConfiguration>
+ AsRef<EthRpcConfiguration>
+ AsRef<WalletConfiguration>,
use_fallback_gas_filler: bool,
) -> Self {
let eth_proxy_binary = AsRef::<EthRpcConfiguration>::as_ref(&context)
.path
@@ -164,6 +167,7 @@ impl ZombienetNode {
connection_string: String::new(),
node_rpc_port: None,
provider: Default::default(),
use_fallback_gas_filler,
}
}
@@ -330,7 +334,12 @@ impl ZombienetNode {
.get_or_try_init(|| async move {
construct_concurrency_limited_provider::<Ethereum, _>(
self.connection_string.as_str(),
FallbackGasFiller::new(u64::MAX, 5_000_000_000, 1_000_000_000),
FallbackGasFiller::new(
u64::MAX,
5_000_000_000,
1_000_000_000,
self.use_fallback_gas_filler,
),
ChainIdFiller::default(), // TODO: use CHAIN_ID constant
NonceFiller::new(self.nonce_manager.clone()),
self.wallet.clone(),
@@ -823,6 +832,7 @@ mod tests {
let mut node = ZombienetNode::new(
context.polkadot_parachain_configuration.path.clone(),
&context,
true,
);
let genesis = context.genesis_configuration.genesis().unwrap().clone();
node.init(genesis).unwrap();
@@ -936,6 +946,7 @@ mod tests {
let node = ZombienetNode::new(
context.polkadot_parachain_configuration.path.clone(),
&context,
true,
);
// Act
@@ -956,6 +967,7 @@ mod tests {
let node = ZombienetNode::new(
context.polkadot_parachain_configuration.path.clone(),
&context,
true,
);
// Act
@@ -4,7 +4,7 @@ use alloy::{
Provider, SendableTx,
fillers::{GasFiller, TxFiller},
},
transports::TransportResult,
transports::{TransportError, TransportResult},
};
// Percentage padding applied to estimated gas (e.g. 120 = 20% padding)
@@ -17,6 +17,7 @@ pub struct FallbackGasFiller {
default_gas_limit: u64,
default_max_fee_per_gas: u128,
default_priority_fee: u128,
use_fallback_gas_filler: bool,
}
impl FallbackGasFiller {
@@ -24,19 +25,41 @@ impl FallbackGasFiller {
default_gas_limit: u64,
default_max_fee_per_gas: u128,
default_priority_fee: u128,
use_fallback_gas_filler: bool,
) -> Self {
Self {
inner: GasFiller,
default_gas_limit,
default_max_fee_per_gas,
default_priority_fee,
use_fallback_gas_filler,
}
}
pub fn with_default_gas_limit(mut self, default_gas_limit: u64) -> Self {
self.default_gas_limit = default_gas_limit;
self
}
pub fn with_default_max_fee_per_gas(mut self, default_max_fee_per_gas: u128) -> Self {
self.default_max_fee_per_gas = default_max_fee_per_gas;
self
}
pub fn with_default_priority_fee(mut self, default_priority_fee: u128) -> Self {
self.default_priority_fee = default_priority_fee;
self
}
pub fn with_use_fallback_gas_filler(mut self, use_fallback_gas_filler: bool) -> Self {
self.use_fallback_gas_filler = use_fallback_gas_filler;
self
}
}
impl Default for FallbackGasFiller {
fn default() -> Self {
FallbackGasFiller::new(25_000_000, 1_000_000_000, 1_000_000_000)
FallbackGasFiller::new(25_000_000, 1_000_000_000, 1_000_000_000, true)
}
}
@@ -64,7 +87,12 @@ where
Ok(fill) => Ok(Some(fill)),
Err(err) => {
tracing::debug!(error = ?err, "Gas Provider Estimation Failed, using fallback");
Ok(None)
if !self.use_fallback_gas_filler {
Err(err)
} else {
Ok(None)
}
}
}
}
@@ -86,13 +114,17 @@ where
}
}
Ok(tx)
} else {
} else if self.use_fallback_gas_filler {
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)
} else {
Err(TransportError::UnsupportedFeature(
"Fallback gas filler is disabled and we're attempting to do a gas estimate on a failing transaction",
))
}
}
}