Compare commits

...

5 Commits

Author SHA1 Message Date
Marian Radu a5f0c8713f Add PVM heap-size and stack-size configuration for resolc 2026-01-15 18:03:57 +02:00
Marian Radu 89ccd72d20 Inject polkavm settings into resolc standard JSON input 2026-01-15 17:24:10 +02:00
Marian Radu 4a4ac7ede6 Update revive compiler dependencies 2026-01-15 17:13:01 +02:00
Omar 94b04c0189 Change the input for the polkadot-omni-node to be a path to chain-spec (#223)
* Change the input for the polkadot-omni-node to be a path to chain-spec

* Remove unneeded dependency
2026-01-14 11:59:21 +00:00
Omar 2d3602aaed Add a retry layer to all providers (#224)
* Add a `ReceiptRetryLayer` for providers

* Fix the retry layer

* Rename the retry layer

* Remove outdated polling function

* Remoe unneeded dependencies
2026-01-13 19:39:47 +00:00
22 changed files with 1585 additions and 1621 deletions
@@ -33,8 +33,8 @@ inputs:
description: "The identifier of the platform to run the tests on (e.g., geth-evm-solc, revive-dev-node-revm-solc)"
required: true
type: string
polkadot-omnichain-node-runtime-path:
description: "The path of the WASM runtime to use with the polkadot-omni-node. This is only required if the polkadot-omni-node is one of the selected platforms."
polkadot-omnichain-node-chain-spec-path:
description: "The path of the chain-spec of the chain we're spawning'. This is only required if the polkadot-omni-node is one of the selected platforms."
required: false
type: string
polkadot-omnichain-node-parachain-id:
@@ -89,10 +89,10 @@ runs:
"${{ inputs['polkadot-omnichain-node-parachain-id'] }}"
)
fi
if [[ -n "${{ inputs['polkadot-omnichain-node-runtime-path'] }}" ]]; then
if [[ -n "${{ inputs['polkadot-omnichain-node-chain-spec-path'] }}" ]]; then
OMNI_ARGS+=(
--polkadot-omni-node.runtime-wasm-path
"${{ inputs['polkadot-omnichain-node-runtime-path'] }}"
--polkadot-omni-node.chain-spec-path
"${{ inputs['polkadot-omnichain-node-chain-spec-path'] }}"
)
fi
+1 -1
View File
@@ -10,7 +10,7 @@ node_modules
*.log
profile.json.gz
workdir
workdir*
!/schema.json
!/dev-genesis.json
Generated
+1257 -1098
View File
File diff suppressed because it is too large Load Diff
+4 -24
View File
@@ -22,6 +22,7 @@ revive-dt-node-pool = { version = "0.1.0", path = "crates/node-pool" }
revive-dt-report = { version = "0.1.0", path = "crates/report" }
revive-dt-solc-binaries = { version = "0.1.0", path = "crates/solc-binaries" }
alloy = { version = "1.4.1", features = ["full", "genesis", "json-rpc"] }
ansi_term = "0.12.1"
anyhow = "1.0"
bson = { version = "2.15.0" }
@@ -72,33 +73,12 @@ indexmap = { version = "2.10.0", default-features = false }
itertools = { version = "0.14.0" }
# revive compiler
revive-solc-json-interface = { git = "https://github.com/paritytech/revive", rev = "3389865af7c3ff6f29a586d82157e8bc573c1a8e" }
revive-common = { git = "https://github.com/paritytech/revive", rev = "3389865af7c3ff6f29a586d82157e8bc573c1a8e" }
revive-differential = { git = "https://github.com/paritytech/revive", rev = "3389865af7c3ff6f29a586d82157e8bc573c1a8e" }
revive-solc-json-interface = { version = "0.5.0" }
revive-common = { version = "0.3.0" }
revive-differential = { version = "0.3.0" }
zombienet-sdk = { git = "https://github.com/paritytech/zombienet-sdk.git", rev = "891f6554354ce466abd496366dbf8b4f82141241" }
[workspace.dependencies.alloy]
version = "1.0.37"
default-features = false
features = [
"json-abi",
"providers",
"provider-ws",
"provider-ipc",
"provider-http",
"provider-debug-api",
"reqwest",
"rpc-types",
"signer-local",
"std",
"network",
"serde",
"rpc-types-eth",
"genesis",
"sol-types",
]
[profile.bench]
inherits = "release"
lto = true
-1
View File
@@ -19,7 +19,6 @@ semver = { workspace = true }
serde = { workspace = true }
schemars = { workspace = true }
strum = { workspace = true }
tokio = { workspace = true, default-features = false, features = ["time"] }
[lints]
workspace = true
-3
View File
@@ -1,3 +0,0 @@
mod poll;
pub use poll::*;
-72
View File
@@ -1,72 +0,0 @@
use std::ops::ControlFlow;
use std::time::Duration;
use anyhow::{Context as _, Result, anyhow};
const EXPONENTIAL_BACKOFF_MAX_WAIT_DURATION: Duration = Duration::from_secs(60);
/// A function that polls for a fallible future for some period of time and errors if it fails to
/// get a result after polling.
///
/// Given a future that returns a [`Result<ControlFlow<O, ()>>`], this function calls the future
/// repeatedly (with some wait period) until the future returns a [`ControlFlow::Break`] or until it
/// returns an [`Err`] in which case the function stops polling and returns the error.
///
/// If the future keeps returning [`ControlFlow::Continue`] and fails to return a [`Break`] within
/// the permitted polling duration then this function returns an [`Err`]
///
/// [`Break`]: ControlFlow::Break
/// [`Continue`]: ControlFlow::Continue
pub async fn poll<F, O>(
polling_duration: Duration,
polling_wait_behavior: PollingWaitBehavior,
mut future: impl FnMut() -> F,
) -> Result<O>
where
F: Future<Output = Result<ControlFlow<O, ()>>>,
{
let mut retries = 0;
let mut total_wait_duration = Duration::ZERO;
let max_allowed_wait_duration = polling_duration;
loop {
if total_wait_duration >= max_allowed_wait_duration {
break Err(anyhow!(
"Polling failed after {} retries and a total of {:?} of wait time",
retries,
total_wait_duration
));
}
match future()
.await
.context("Polled future returned an error during polling loop")?
{
ControlFlow::Continue(()) => {
let next_wait_duration = match polling_wait_behavior {
PollingWaitBehavior::Constant(duration) => duration,
PollingWaitBehavior::ExponentialBackoff => {
Duration::from_secs(2u64.pow(retries))
.min(EXPONENTIAL_BACKOFF_MAX_WAIT_DURATION)
}
};
let next_wait_duration =
next_wait_duration.min(max_allowed_wait_duration - total_wait_duration);
total_wait_duration += next_wait_duration;
retries += 1;
tokio::time::sleep(next_wait_duration).await;
}
ControlFlow::Break(output) => {
break Ok(output);
}
}
}
}
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Default)]
pub enum PollingWaitBehavior {
Constant(Duration),
#[default]
ExponentialBackoff,
}
-1
View File
@@ -3,7 +3,6 @@
pub mod cached_fs;
pub mod fs;
pub mod futures;
pub mod iterators;
pub mod macros;
pub mod types;
+65 -23
View File
@@ -12,9 +12,13 @@ use dashmap::DashMap;
use revive_dt_common::types::VersionOrRequirement;
use revive_dt_config::{ResolcConfiguration, SolcConfiguration, WorkingDirectoryConfiguration};
use revive_solc_json_interface::{
SolcStandardJsonInput, SolcStandardJsonInputLanguage, SolcStandardJsonInputSettings,
SolcStandardJsonInputSettingsOptimizer, SolcStandardJsonInputSettingsSelection,
SolcStandardJsonOutput,
PolkaVMDefaultHeapMemorySize, PolkaVMDefaultStackMemorySize, SolcStandardJsonInput,
SolcStandardJsonInputLanguage, SolcStandardJsonInputSettings,
SolcStandardJsonInputSettingsLibraries, SolcStandardJsonInputSettingsMetadata,
SolcStandardJsonInputSettingsOptimizer, SolcStandardJsonInputSettingsPolkaVM,
SolcStandardJsonInputSettingsPolkaVMMemory, SolcStandardJsonInputSettingsSelection,
SolcStandardJsonOutput, standard_json::input::settings::optimizer::Optimizer,
standard_json::input::settings::optimizer::details::Details,
};
use tracing::{Span, field::display};
@@ -25,6 +29,7 @@ use crate::{
use alloy::json_abi::JsonAbi;
use anyhow::{Context as _, Result};
use semver::Version;
use std::collections::BTreeSet;
use tokio::{io::AsyncWriteExt, process::Command as AsyncCommand};
/// A wrapper around the `resolc` binary, emitting PVM-compatible bytecode.
@@ -37,6 +42,10 @@ struct ResolcInner {
solc: Solc,
/// Path to the `resolc` executable
resolc_path: PathBuf,
/// The PVM heap size in bytes.
pvm_heap_size: u32,
/// The PVM stack size in bytes.
pvm_stack_size: u32,
}
impl Resolc {
@@ -63,10 +72,35 @@ impl Resolc {
Self(Arc::new(ResolcInner {
solc,
resolc_path: resolc_configuration.path.clone(),
pvm_heap_size: resolc_configuration
.heap_size
.unwrap_or(PolkaVMDefaultHeapMemorySize),
pvm_stack_size: resolc_configuration
.stack_size
.unwrap_or(PolkaVMDefaultStackMemorySize),
}))
})
.clone())
}
fn polkavm_settings(&self) -> SolcStandardJsonInputSettingsPolkaVM {
SolcStandardJsonInputSettingsPolkaVM::new(
Some(SolcStandardJsonInputSettingsPolkaVMMemory::new(
Some(self.0.pvm_heap_size),
Some(self.0.pvm_stack_size),
)),
false,
)
}
fn inject_polkavm_settings(&self, input: &SolcStandardJsonInput) -> Result<serde_json::Value> {
let mut input_value = serde_json::to_value(input)
.context("Failed to serialize Standard JSON input for resolc")?;
if let Some(settings) = input_value.get_mut("settings") {
settings["polkavm"] = serde_json::to_value(self.polkavm_settings()).unwrap();
}
Ok(input_value)
}
}
impl SolidityCompiler for Resolc {
@@ -121,8 +155,8 @@ impl SolidityCompiler for Resolc {
.collect(),
settings: SolcStandardJsonInputSettings {
evm_version,
libraries: Some(
libraries
libraries: SolcStandardJsonInputSettingsLibraries {
inner: libraries
.into_iter()
.map(|(source_code, libraries_map)| {
(
@@ -136,23 +170,29 @@ impl SolidityCompiler for Resolc {
)
})
.collect(),
),
remappings: None,
output_selection: Some(SolcStandardJsonInputSettingsSelection::new_required()),
},
remappings: BTreeSet::<String>::new(),
output_selection: SolcStandardJsonInputSettingsSelection::new_required(),
via_ir: Some(true),
optimizer: SolcStandardJsonInputSettingsOptimizer::new(
optimization
.unwrap_or(ModeOptimizerSetting::M0)
.optimizations_enabled(),
None,
&Version::new(0, 0, 0),
false,
Optimizer::default_mode(),
Details::disabled(&Version::new(0, 0, 0)),
),
metadata: None,
polkavm: None,
polkavm: self.polkavm_settings(),
metadata: SolcStandardJsonInputSettingsMetadata::default(),
detect_missing_libraries: false,
},
};
Span::current().record("json_in", display(serde_json::to_string(&input).unwrap()));
// Manually inject polkavm settings since it's marked skip_serializing in the upstream crate
let std_input_json = self.inject_polkavm_settings(&input)?;
Span::current().record(
"json_in",
display(serde_json::to_string(&std_input_json).unwrap()),
);
let path = &self.0.resolc_path;
let mut command = AsyncCommand::new(path);
@@ -181,8 +221,9 @@ impl SolidityCompiler for Resolc {
.with_context(|| format!("Failed to spawn resolc at {}", path.display()))?;
let stdin_pipe = child.stdin.as_mut().expect("stdin must be piped");
let serialized_input = serde_json::to_vec(&input)
let serialized_input = serde_json::to_vec(&std_input_json)
.context("Failed to serialize Standard JSON input for resolc")?;
stdin_pipe
.write_all(&serialized_input)
.await
@@ -228,7 +269,7 @@ impl SolidityCompiler for Resolc {
// Detecting if the compiler output contained errors and reporting them through logs and
// errors instead of returning the compiler output that might contain errors.
for error in parsed.errors.iter().flatten() {
for error in parsed.errors.iter() {
if error.severity == "error" {
tracing::error!(
?error,
@@ -240,12 +281,12 @@ impl SolidityCompiler for Resolc {
}
}
let Some(contracts) = parsed.contracts else {
if parsed.contracts.is_empty() {
anyhow::bail!("Unexpected error - resolc output doesn't have a contracts section");
};
}
let mut compiler_output = CompilerOutput::default();
for (source_path, contracts) in contracts.into_iter() {
for (source_path, contracts) in parsed.contracts.into_iter() {
let src_for_msg = source_path.clone();
let source_path = PathBuf::from(source_path)
.canonicalize()
@@ -258,10 +299,11 @@ impl SolidityCompiler for Resolc {
.and_then(|evm| evm.bytecode.clone())
.context("Unexpected - Contract compiled with resolc has no bytecode")?;
let abi = {
let metadata = contract_information
.metadata
.as_ref()
.context("No metadata found for the contract")?;
let metadata = &contract_information.metadata;
if metadata.is_null() {
anyhow::bail!("No metadata found for the contract");
}
let solc_metadata_str = match metadata {
serde_json::Value::String(solc_metadata_str) => {
solc_metadata_str.as_str()
+15 -5
View File
@@ -800,6 +800,17 @@ pub struct ResolcConfiguration {
/// provided in the user's $PATH.
#[clap(id = "resolc.path", long = "resolc.path", default_value = "resolc")]
pub path: PathBuf,
/// Specifies the PVM heap size in bytes.
///
/// If unspecified, the revive compiler default is used
#[clap(id = "resolc.heap-size", long = "resolc.heap-size")]
pub heap_size: Option<u32>,
/// Specifies the PVM stack size in bytes.
///
/// If unspecified, the revive compiler default is used
#[clap(id = "resolc.stack-size", long = "resolc.stack-size")]
pub stack_size: Option<u32>,
}
/// A set of configuration parameters for Polkadot Parachain.
@@ -943,13 +954,12 @@ pub struct PolkadotOmnichainNodeConfiguration {
)]
pub block_time: Duration,
/// The path of the WASM runtime to use for the polkadot-omni-node. This argument is required if
/// the polkadot-omni-node is one of the selected platforms for running the tests or benchmarks.
/// The path of the chainspec of the chain that we're spawning
#[clap(
id = "polkadot-omni-node.runtime-wasm-path",
long = "polkadot-omni-node.runtime-wasm-path"
id = "polkadot-omni-node.chain-spec-path",
long = "polkadot-omni-node.chain-spec-path"
)]
pub runtime_wasm_path: Option<PathBuf>,
pub chain_spec_path: Option<PathBuf>,
/// The ID of the parachain that the polkadot-omni-node will spawn. This argument is required if
/// the polkadot-omni-node is one of the selected platforms for running the tests or benchmarks.
@@ -409,7 +409,6 @@ where
.handle_function_call_execution(step, deployment_receipts)
.await
.context("Failed to handle the function call execution")?;
tracing::Span::current().record("block_number", execution_receipt.block_number);
let tracing_result = self
.handle_function_call_call_frame_tracing(execution_receipt.transaction_hash)
.await
+2 -10
View File
@@ -486,13 +486,9 @@ impl Platform for PolkadotOmniNodePolkavmResolcPlatform {
let wallet = AsRef::<WalletConfiguration>::as_ref(&context).wallet();
PolkadotOmnichainNode::node_genesis(
&polkadot_omnichain_node_configuration.path,
&wallet,
polkadot_omnichain_node_configuration
.parachain_id
.context("No parachain id found in the configuration of the polkadot-omni-node")?,
polkadot_omnichain_node_configuration
.runtime_wasm_path
.chain_spec_path
.as_ref()
.context("No WASM runtime path found in the polkadot-omni-node configuration")?,
)
@@ -550,13 +546,9 @@ impl Platform for PolkadotOmniNodeRevmSolcPlatform {
let wallet = AsRef::<WalletConfiguration>::as_ref(&context).wallet();
PolkadotOmnichainNode::node_genesis(
&polkadot_omnichain_node_configuration.path,
&wallet,
polkadot_omnichain_node_configuration
.parachain_id
.context("No parachain id found in the configuration of the polkadot-omni-node")?,
polkadot_omnichain_node_configuration
.runtime_wasm_path
.chain_spec_path
.as_ref()
.context("No WASM runtime path found in the polkadot-omni-node configuration")?,
)
-1
View File
@@ -31,7 +31,6 @@ serde_yaml_ng = { workspace = true }
sp-core = { workspace = true }
sp-runtime = { workspace = true }
subxt = { workspace = true }
temp-dir = { workspace = true }
zombienet-sdk = { workspace = true }
[dev-dependencies]
+13 -92
View File
@@ -3,7 +3,6 @@
use std::{
fs::{File, create_dir_all, remove_dir_all},
io::Read,
ops::ControlFlow,
path::PathBuf,
pin::Pin,
process::{Command, Stdio},
@@ -35,12 +34,9 @@ use anyhow::Context as _;
use futures::{FutureExt, Stream, StreamExt};
use revive_common::EVMVersion;
use tokio::sync::OnceCell;
use tracing::{Instrument, error, instrument};
use tracing::{error, instrument};
use revive_dt_common::{
fs::clear_directory,
futures::{PollingWaitBehavior, poll},
};
use revive_dt_common::fs::clear_directory;
use revive_dt_config::*;
use revive_dt_format::traits::ResolverApi;
use revive_dt_node_interaction::EthereumNode;
@@ -90,12 +86,6 @@ impl GethNode {
const READY_MARKER: &str = "IPC endpoint opened";
const ERROR_MARKER: &str = "Fatal:";
const TRANSACTION_INDEXING_ERROR: &str = "transaction indexing is in progress";
const TRANSACTION_TRACING_ERROR: &str = "historical state not available in path scheme yet";
const RECEIPT_POLLING_DURATION: Duration = Duration::from_secs(5 * 60);
const TRACE_POLLING_DURATION: Duration = Duration::from_secs(60);
pub fn new(
context: impl AsRef<WorkingDirectoryConfiguration>
+ AsRef<WalletConfiguration>
@@ -341,62 +331,15 @@ impl EthereumNode for GethNode {
transaction: TransactionRequest,
) -> Pin<Box<dyn Future<Output = anyhow::Result<TransactionReceipt>> + '_>> {
Box::pin(async move {
let provider = self
.provider()
self.provider()
.await
.context("Failed to create provider for transaction submission")?;
let pending_transaction = provider
.context("Failed to create provider for transaction submission")?
.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
// to get. You can find more information on this in the following GH issue in geth
// https://github.com/ethereum/go-ethereum/issues/28877. To summarize what's going on,
// before we can get the receipt of the transaction it needs to have been indexed by the
// node's indexer. Just because the transaction has been confirmed it doesn't mean that it
// has been indexed. When we call alloy's `get_receipt` it checks if the transaction was
// confirmed. If it has been, then it will call `eth_getTransactionReceipt` method which
// _might_ return the above error if the tx has not yet been indexed yet. So, we need to
// implement a retry mechanism for the receipt to keep retrying to get it until it
// eventually works, but we only do that if the error we get back is the "transaction
// indexing is in progress" error or if the receipt is None.
//
// Getting the transaction indexed and taking a receipt can take a long time especially when
// a lot of transactions are being submitted to the node. Thus, while initially we only
// 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.
poll(
Self::RECEIPT_POLLING_DURATION,
PollingWaitBehavior::Constant(Duration::from_millis(200)),
move || {
let provider = provider.clone();
async move {
match provider.get_transaction_receipt(transaction_hash).await {
Ok(Some(receipt)) => Ok(ControlFlow::Break(receipt)),
Ok(None) => Ok(ControlFlow::Continue(())),
Err(error) => {
let error_string = error.to_string();
match error_string.contains(Self::TRANSACTION_INDEXING_ERROR) {
true => Ok(ControlFlow::Continue(())),
false => Err(error.into()),
}
}
}
}
},
)
.instrument(tracing::info_span!(
"Awaiting transaction receipt",
?transaction_hash
))
.await
.context("Encountered an error when submitting a transaction")?
.get_receipt()
.await
.context("Failed to get the receipt for the transaction")
})
}
@@ -407,34 +350,12 @@ impl EthereumNode for GethNode {
trace_options: GethDebugTracingOptions,
) -> Pin<Box<dyn Future<Output = anyhow::Result<GethTrace>> + '_>> {
Box::pin(async move {
let provider = self
.provider()
self.provider()
.await
.context("Failed to create provider for tracing")?;
poll(
Self::TRACE_POLLING_DURATION,
PollingWaitBehavior::Constant(Duration::from_millis(200)),
move || {
let provider = provider.clone();
let trace_options = trace_options.clone();
async move {
match provider
.debug_trace_transaction(tx_hash, trace_options)
.await
{
Ok(trace) => Ok(ControlFlow::Break(trace)),
Err(error) => {
let error_string = error.to_string();
match error_string.contains(Self::TRANSACTION_TRACING_ERROR) {
true => Ok(ControlFlow::Continue(())),
false => Err(error.into()),
}
}
}
}
},
)
.await
.context("Failed to create provider for tracing")?
.debug_trace_transaction(tx_hash, trace_options)
.await
.context("Failed to get the transaction trace")
})
}
@@ -12,7 +12,6 @@ use std::{
collections::{BTreeMap, HashSet},
fs::{File, create_dir_all},
io::Read,
ops::ControlFlow,
path::PathBuf,
pin::Pin,
process::{Command, Stdio},
@@ -48,12 +47,9 @@ use revive_common::EVMVersion;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use serde_with::serde_as;
use tokio::sync::OnceCell;
use tracing::{Instrument, info, instrument};
use tracing::{info, instrument};
use revive_dt_common::{
fs::clear_directory,
futures::{PollingWaitBehavior, poll},
};
use revive_dt_common::fs::clear_directory;
use revive_dt_config::*;
use revive_dt_format::traits::ResolverApi;
use revive_dt_node_interaction::EthereumNode;
@@ -116,12 +112,6 @@ impl LighthouseGethNode {
const CONFIG_FILE_NAME: &str = "config.yaml";
const TRANSACTION_INDEXING_ERROR: &str = "transaction indexing is in progress";
const TRANSACTION_TRACING_ERROR: &str = "historical state not available in path scheme yet";
const RECEIPT_POLLING_DURATION: Duration = Duration::from_secs(5 * 60);
const TRACE_POLLING_DURATION: Duration = Duration::from_secs(60);
const VALIDATOR_MNEMONIC: &str = "giant issue aisle success illegal bike spike question tent bar rely arctic volcano long crawl hungry vocal artwork sniff fantasy very lucky have athlete";
pub fn new(
@@ -481,73 +471,6 @@ impl LighthouseGethNode {
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
.inspect_err(|err| {
tracing::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 to get. You can find more information on this in the following GH issue in geth
// https://github.com/ethereum/go-ethereum/issues/28877. To summarize what's going on,
// before we can get the receipt of the transaction it needs to have been indexed by the
// node's indexer. Just because the transaction has been confirmed it doesn't mean that
// it has been indexed. When we call alloy's `get_receipt` it checks if the transaction
// was confirmed. If it has been, then it will call `eth_getTransactionReceipt` method
// which _might_ return the above error if the tx has not yet been indexed yet. So, we
// need to implement a retry mechanism for the receipt to keep retrying to get it until
// it eventually works, but we only do that if the error we get back is the "transaction
// indexing is in progress" error or if the receipt is None.
//
// Getting the transaction indexed and taking a receipt can take a long time especially
// when a lot of transactions are being submitted to the node. Thus, while initially we
// only 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.
poll(
Self::RECEIPT_POLLING_DURATION,
PollingWaitBehavior::Constant(Duration::from_millis(500)),
move || {
let provider = provider.clone();
async move {
match provider.get_transaction_receipt(transaction_hash).await {
Ok(Some(receipt)) => Ok(ControlFlow::Break(receipt)),
Ok(None) => Ok(ControlFlow::Continue(())),
Err(error) => {
let error_string = error.to_string();
match error_string.contains(Self::TRANSACTION_INDEXING_ERROR) {
true => Ok(ControlFlow::Continue(())),
false => Err(error.into()),
}
}
}
}
},
)
.instrument(tracing::info_span!(
"Awaiting transaction receipt",
?transaction_hash
))
.await
})
}
pub fn node_genesis(mut genesis: Genesis, wallet: &EthereumWallet) -> Genesis {
for signer_address in NetworkWallet::<Ethereum>::signer_addresses(&wallet) {
genesis
@@ -626,11 +549,15 @@ impl EthereumNode for LighthouseGethNode {
transaction: TransactionRequest,
) -> Pin<Box<dyn Future<Output = anyhow::Result<TransactionReceipt>> + '_>> {
Box::pin(async move {
let provider = self
.http_provider()
self.provider()
.await
.context("Failed to create provider for transaction execution")?;
Self::internal_execute_transaction(transaction, provider).await
.context("Failed to create provider for transaction submission")?
.send_transaction(transaction)
.await
.context("Encountered an error when submitting a transaction")?
.get_receipt()
.await
.context("Failed to get the receipt for the transaction")
})
}
@@ -641,35 +568,12 @@ impl EthereumNode for LighthouseGethNode {
trace_options: GethDebugTracingOptions,
) -> Pin<Box<dyn Future<Output = anyhow::Result<GethTrace>> + '_>> {
Box::pin(async move {
let provider = Arc::new(
self.http_provider()
.await
.context("Failed to create provider for tracing")?,
);
poll(
Self::TRACE_POLLING_DURATION,
PollingWaitBehavior::Constant(Duration::from_millis(200)),
move || {
let provider = provider.clone();
let trace_options = trace_options.clone();
async move {
match provider
.debug_trace_transaction(tx_hash, trace_options)
.await
{
Ok(trace) => Ok(ControlFlow::Break(trace)),
Err(error) => {
let error_string = error.to_string();
match error_string.contains(Self::TRANSACTION_TRACING_ERROR) {
true => Ok(ControlFlow::Continue(())),
false => Err(error.into()),
}
}
}
}
},
)
.await
self.provider()
.await
.context("Failed to create provider for tracing")?
.debug_trace_transaction(tx_hash, trace_options)
.await
.context("Failed to get the transaction trace")
})
}
@@ -42,7 +42,6 @@ use revive_dt_report::{
EthereumMinedBlockInformation, MinedBlockInformation, SubstrateMinedBlockInformation,
};
use subxt::{OnlineClient, SubstrateConfig};
use temp_dir::TempDir;
use tokio::sync::OnceCell;
use tracing::{instrument, trace};
@@ -50,10 +49,7 @@ use crate::{
Node,
constants::INITIAL_BALANCE,
helpers::{Process, ProcessReadinessWaitBehavior},
provider_utils::{
ConcreteProvider, FallbackGasFiller, construct_concurrency_limited_provider,
execute_transaction,
},
provider_utils::{ConcreteProvider, FallbackGasFiller, construct_concurrency_limited_provider},
};
static NODE_COUNT: AtomicU32 = AtomicU32::new(0);
@@ -73,7 +69,7 @@ pub struct PolkadotOmnichainNode {
/// The path of the eth-rpc binary.
eth_rpc_binary_path: PathBuf,
/// The path of the runtime's WASM that this node will be spawned with.
runtime_wasm_path: Option<PathBuf>,
chain_spec_path: Option<PathBuf>,
/// The path of the base directory which contains all of the stored data for this node.
base_directory_path: PathBuf,
/// The path of the logs directory which contains all of the stored logs.
@@ -147,8 +143,8 @@ impl PolkadotOmnichainNode {
.path
.to_path_buf(),
eth_rpc_binary_path: eth_rpc_path.to_path_buf(),
runtime_wasm_path: polkadot_omnichain_node_configuration
.runtime_wasm_path
chain_spec_path: polkadot_omnichain_node_configuration
.chain_spec_path
.clone(),
base_directory_path: base_directory,
logs_directory_path: logs_directory,
@@ -180,10 +176,8 @@ impl PolkadotOmnichainNode {
let template_chainspec_path = self.base_directory_path.join(Self::CHAIN_SPEC_JSON_FILE);
let chainspec_json = Self::node_genesis(
&self.polkadot_omnichain_node_binary_path,
&self.wallet,
self.parachain_id.context("No parachain id provided")?,
self.runtime_wasm_path
self.chain_spec_path
.as_ref()
.context("No runtime path provided")?,
)
@@ -202,7 +196,7 @@ impl PolkadotOmnichainNode {
fn spawn_process(&mut self) -> anyhow::Result<()> {
// Error out if the runtime's path or the parachain id are not set which means that the
// arguments we require were not provided.
self.runtime_wasm_path
self.chain_spec_path
.as_ref()
.context("No WASM path provided for the runtime")?;
self.parachain_id
@@ -361,40 +355,11 @@ impl PolkadotOmnichainNode {
}
pub fn node_genesis(
node_path: &Path,
wallet: &EthereumWallet,
parachain_id: usize,
runtime_wasm_path: &Path,
chain_spec_path: &Path,
) -> anyhow::Result<serde_json::Value> {
let tempdir = TempDir::new().context("Failed to create a temporary directory")?;
let unmodified_chainspec_path = tempdir.path().join("chainspec.json");
let output = Command::new(node_path)
.arg("chain-spec-builder")
.arg("-c")
.arg(unmodified_chainspec_path.as_path())
.arg("create")
.arg("--para-id")
.arg(parachain_id.to_string())
.arg("--relay-chain")
.arg("dontcare")
.arg("--runtime")
.arg(runtime_wasm_path)
.arg("named-preset")
.arg("development")
.env_remove("RUST_LOG")
.output()
.context("Failed to export the chain-spec")?;
if !output.status.success() {
anyhow::bail!(
"Exporting chainspec from polkadot-omni-node failed: {}",
String::from_utf8_lossy(&output.stderr)
);
}
let unmodified_chainspec_file = File::open(unmodified_chainspec_path.as_path())
.context("Failed to open the unmodified chainspec file")?;
let unmodified_chainspec_file =
File::open(chain_spec_path).context("Failed to open the unmodified chainspec file")?;
let mut chainspec_json =
serde_json::from_reader::<_, serde_json::Value>(&unmodified_chainspec_file)
.context("Failed to read the unmodified chainspec JSON")?;
@@ -464,11 +429,15 @@ impl EthereumNode for PolkadotOmnichainNode {
transaction: TransactionRequest,
) -> Pin<Box<dyn Future<Output = anyhow::Result<TransactionReceipt>> + '_>> {
Box::pin(async move {
let provider = self
.provider()
self.provider()
.await
.context("Failed to create the provider")?;
execute_transaction(provider, transaction).await
.context("Failed to create provider for transaction submission")?
.send_transaction(transaction)
.await
.context("Encountered an error when submitting a transaction")?
.get_receipt()
.await
.context("Failed to get the receipt for the transaction")
})
}
@@ -49,10 +49,7 @@ use crate::{
Node,
constants::INITIAL_BALANCE,
helpers::{Process, ProcessReadinessWaitBehavior},
provider_utils::{
ConcreteProvider, FallbackGasFiller, construct_concurrency_limited_provider,
execute_transaction,
},
provider_utils::{ConcreteProvider, FallbackGasFiller, construct_concurrency_limited_provider},
};
static NODE_COUNT: AtomicU32 = AtomicU32::new(0);
@@ -434,11 +431,15 @@ impl EthereumNode for SubstrateNode {
transaction: TransactionRequest,
) -> Pin<Box<dyn Future<Output = anyhow::Result<TransactionReceipt>> + '_>> {
Box::pin(async move {
let provider = self
.provider()
self.provider()
.await
.context("Failed to create the provider")?;
execute_transaction(provider, transaction).await
.context("Failed to create provider for transaction submission")?
.send_transaction(transaction)
.await
.context("Encountered an error when submitting a transaction")?
.get_receipt()
.await
.context("Failed to get the receipt for the transaction")
})
}
@@ -76,10 +76,7 @@ use crate::{
Node,
constants::INITIAL_BALANCE,
helpers::{Process, ProcessReadinessWaitBehavior},
provider_utils::{
ConcreteProvider, FallbackGasFiller, construct_concurrency_limited_provider,
execute_transaction,
},
provider_utils::{ConcreteProvider, FallbackGasFiller, construct_concurrency_limited_provider},
};
static NODE_COUNT: AtomicU32 = AtomicU32::new(0);
@@ -433,14 +430,18 @@ impl EthereumNode for ZombienetNode {
fn execute_transaction(
&self,
transaction: alloy::rpc::types::TransactionRequest,
transaction: TransactionRequest,
) -> Pin<Box<dyn Future<Output = anyhow::Result<TransactionReceipt>> + '_>> {
Box::pin(async move {
let provider = self
.provider()
self.provider()
.await
.context("Failed to create the provider")?;
execute_transaction(provider, transaction).await
.context("Failed to create provider for transaction submission")?
.send_transaction(transaction)
.await
.context("Encountered an error when submitting a transaction")?
.get_receipt()
.await
.context("Failed to get the receipt for the transaction")
})
}
@@ -1,5 +1,3 @@
use std::{borrow::Cow, fmt::Display};
use alloy::{
eips::BlockNumberOrTag,
network::{Network, TransactionBuilder},
@@ -111,28 +109,23 @@ where
},
state_overrides: Default::default(),
block_overrides: Default::default(),
tx_index: Default::default(),
},
)
.await?
.try_into_call_frame()
.map_err(|err| {
RpcError::LocalUsageError(
FallbackGasFillerError::new(format!(
"Expected a callframe trace, but got: {err:?}"
))
.boxed(),
RpcError::local_usage_str(
format!("Expected a callframe trace, but got: {err:?}").as_str(),
)
})?;
let gas_used = u64::try_from(trace.gas_used).map_err(|_| {
RpcError::LocalUsageError(
FallbackGasFillerError::new(
"Transaction trace returned a value of gas used that exceeds u64",
)
.boxed(),
RpcError::local_usage_str(
"Transaction trace returned a value of gas used that exceeds u64",
)
})?;
let gas_limit = gas_used.saturating_mul(120) / 100;
let gas_limit = gas_used.saturating_mul(2);
if let Some(gas_price) = tx.gas_price() {
return Ok(GasFillable::Legacy {
@@ -174,24 +167,3 @@ impl Default for FallbackGasFiller {
Self::new()
}
}
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
struct FallbackGasFillerError(Cow<'static, str>);
impl FallbackGasFillerError {
pub fn new(string: impl Into<Cow<'static, str>>) -> Self {
Self(string.into())
}
pub fn boxed(self) -> Box<Self> {
Box::new(self)
}
}
impl Display for FallbackGasFillerError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
Display::fmt(&self.0, f)
}
}
impl std::error::Error for FallbackGasFillerError {}
+2
View File
@@ -1,7 +1,9 @@
mod concurrency_limiter;
mod fallback_gas_filler;
mod provider;
mod receipt_retry_layer;
pub use concurrency_limiter::*;
pub use fallback_gas_filler::*;
pub use provider::*;
pub use receipt_retry_layer::*;
+5 -73
View File
@@ -1,18 +1,16 @@
use std::{ops::ControlFlow, sync::LazyLock, time::Duration};
use std::sync::LazyLock;
use alloy::{
network::{Ethereum, Network, NetworkWallet, TransactionBuilder4844},
network::{Network, NetworkWallet, TransactionBuilder4844},
providers::{
Identity, PendingTransactionBuilder, Provider, ProviderBuilder, RootProvider,
Identity, ProviderBuilder, RootProvider,
fillers::{ChainIdFiller, FillProvider, JoinFill, NonceFiller, TxFiller, WalletFiller},
},
rpc::client::ClientBuilder,
};
use anyhow::{Context, Result};
use revive_dt_common::futures::{PollingWaitBehavior, poll};
use tracing::{Instrument, debug, info, info_span};
use crate::provider_utils::{ConcurrencyLimiterLayer, FallbackGasFiller};
use crate::provider_utils::{ConcurrencyLimiterLayer, FallbackGasFiller, RetryLayer};
pub type ConcreteProvider<N, W> = FillProvider<
JoinFill<
@@ -48,6 +46,7 @@ where
let client = ClientBuilder::default()
.layer(GLOBAL_CONCURRENCY_LIMITER_LAYER.clone())
.layer(RetryLayer::default())
.connect(rpc_url)
.await
.context("Failed to construct the RPC client")?;
@@ -63,70 +62,3 @@ where
Ok(provider)
}
pub async fn execute_transaction<N, W>(
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>,
{
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 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");
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();
async move {
match provider.get_transaction_receipt(tx_hash).await {
Ok(Some(receipt)) => {
info!("Found the transaction receipt");
Ok(ControlFlow::Break(receipt))
}
_ => Ok(ControlFlow::Continue(())),
}
}
},
)
.instrument(info_span!("Polling for receipt", %tx_hash))
.await
.context(format!("Polling for receipt failed for {tx_hash}"))
}
@@ -0,0 +1,158 @@
use std::time::Duration;
use alloy::{
network::{AnyNetwork, Network},
rpc::json_rpc::{RequestPacket, ResponsePacket},
transports::{TransportError, TransportErrorKind, TransportFut},
};
use tokio::time::{interval, timeout};
use tower::{Layer, Service};
/// A layer that allows for automatic retries for getting the receipt.
///
/// There are certain cases where getting the receipt of a committed transaction might fail. In Geth
/// this can happen if the transaction has been committed to the ledger but has not been indexed, in
/// the substrate and revive stack it can also happen for other reasons.
///
/// Therefore, just because the first attempt to get the receipt (after transaction confirmation)
/// has failed it doesn't mean that it will continue to fail. This layer can be added to any alloy
/// provider to allow the provider to retry getting the receipt for some period of time before it
/// considers that a timeout. It attempts to poll for the receipt for the `polling_duration` with an
/// interval of `polling_interval` between each poll. If by the end of the `polling_duration` it was
/// not able to get the receipt successfully then this is considered to be a timeout.
///
/// Additionally, this layer allows for retries for other rpc methods such as all tracing methods.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct RetryLayer {
/// The amount of time to keep polling for the receipt before considering it a timeout.
polling_duration: Duration,
/// The interval of time to wait between each poll for the receipt.
polling_interval: Duration,
}
impl RetryLayer {
pub fn new(polling_duration: Duration, polling_interval: Duration) -> Self {
Self {
polling_duration,
polling_interval,
}
}
pub fn with_polling_duration(mut self, polling_duration: Duration) -> Self {
self.polling_duration = polling_duration;
self
}
pub fn with_polling_interval(mut self, polling_interval: Duration) -> Self {
self.polling_interval = polling_interval;
self
}
}
impl Default for RetryLayer {
fn default() -> Self {
Self {
polling_duration: Duration::from_secs(90),
polling_interval: Duration::from_millis(500),
}
}
}
impl<S> Layer<S> for RetryLayer {
type Service = RetryService<S>;
fn layer(&self, inner: S) -> Self::Service {
RetryService {
service: inner,
polling_duration: self.polling_duration,
polling_interval: self.polling_interval,
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct RetryService<S> {
/// The internal service.
service: S,
/// The amount of time to keep polling for the receipt before considering it a timeout.
polling_duration: Duration,
/// The interval of time to wait between each poll for the receipt.
polling_interval: Duration,
}
impl<S> Service<RequestPacket> for RetryService<S>
where
S: Service<RequestPacket, Future = TransportFut<'static>, Error = TransportError>
+ Send
+ 'static
+ Clone,
{
type Response = ResponsePacket;
type Error = TransportError;
type Future = TransportFut<'static>;
fn poll_ready(
&mut self,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<Result<(), Self::Error>> {
self.service.poll_ready(cx)
}
#[allow(clippy::nonminimal_bool)]
fn call(&mut self, req: RequestPacket) -> Self::Future {
type ReceiptOutput = <AnyNetwork as Network>::ReceiptResponse;
let mut service = self.service.clone();
let polling_interval = self.polling_interval;
let polling_duration = self.polling_duration;
Box::pin(async move {
let request = req.as_single().ok_or_else(|| {
TransportErrorKind::custom_str("Retry layer doesn't support batch requests")
})?;
let method = request.method();
let requires_retries = method == "eth_getTransactionReceipt"
|| (method.contains("debug") && method.contains("trace"));
if !requires_retries {
return service.call(req).await;
}
timeout(polling_duration, async {
let mut interval = interval(polling_interval);
loop {
interval.tick().await;
let Ok(resp) = service.call(req.clone()).await else {
continue;
};
let response = resp.as_single().expect("Can't fail");
if response.is_error() {
continue;
}
if method == "eth_getTransactionReceipt"
&& response
.payload()
.clone()
.deserialize_success::<ReceiptOutput>()
.ok()
.and_then(|resp| resp.try_into_success().ok())
.is_some()
|| method != "eth_getTransactionReceipt"
{
return resp;
} else {
continue;
}
}
})
.await
.map_err(|_| TransportErrorKind::custom_str("Timeout when retrying request"))
})
}
}