Compare commits

..

26 Commits

Author SHA1 Message Date
yongkangc
7477c978e4 fix: remove dead code using non-existent RocksDBProvider::clear method
The stage/drop.rs was trying to call rocksdb.clear::<T>() but this
method doesn't exist on RocksDBProvider. Remove the dead code blocks
that would have caused compilation failures with the edge feature.
2026-01-16 09:24:27 +00:00
yongkangc
766eabce9a feat(provider): add RocksDBProvider::clear method for table cleanup
Add clear() method to RocksDBProvider and its stub implementation to
support clearing all entries from a RocksDB table. This is needed by
the stage drop CLI command to clear RocksDB history tables.

- Adds clear<T: Table>() method that iterates and deletes all entries
- Stub implementation returns Ok(()) since stub behaves as empty DB
2026-01-16 09:15:10 +00:00
yongkangc
7c1744a898 fix: update CLI commands and test utilities for RocksDB support
Update various CLI commands and test utilities for RocksDB:
- Stage drop command handles RocksDB table clearing
- Stage dump commands pass storage settings
- Test utilities provide RocksDB stubs
- Add PendingRocksDBBatches export
2026-01-16 08:37:04 +00:00
yongkangc
f775cc9b50 docs: update CLI documentation for --storage.rocksdb flag 2026-01-15 10:09:45 +00:00
yongkangc
bba7821495 fix: clippy warnings (doc_markdown, unreachable_pub, for_kv_map) 2026-01-15 10:09:45 +00:00
yongkangc
9ef853d7a1 feat(cli): add --storage.rocksdb flag for RocksDB history indices
Add a single CLI flag to enable RocksDB for all history tables:
- --storage.rocksdb enables AccountsHistory, StoragesHistory, and
  TransactionHashNumbers to be stored in RocksDB instead of MDBX

Also adds:
- `reth db settings set` subcommands for individual table control
- Updated CLI documentation
2026-01-15 10:09:45 +00:00
yongkangc
ff75aa92aa feat(storage): RocksDB support for genesis init and tx lookup stage
Add RocksDB support for:
- Genesis history initialization via EitherWriter
- TransactionLookup stage unwind operations
- RocksDB provider append_*_history_shard methods
- Metrics for RocksDB write operations
2026-01-15 10:09:26 +00:00
yongkangc
126621c5d9 feat(stages): implement RocksDB unwind for index history stages
Add RocksDB unwind support for IndexAccountHistory and IndexStorageHistory stages:
- Create RocksDB batch during unwind when storage settings enabled
- Use EitherWriter for history shard operations
- Register batches with pending_batches for atomic commit
2026-01-15 10:08:49 +00:00
yongkangc
fc776db5d8 feat(stages): add RocksDB helper functions for stage operations
Add utility functions for RocksDB integration in stages:
- make_rocksdb_provider() - creates RocksDB provider from UnifiedStorageWriter
- make_rocksdb_batch_arg() - creates RocksDB batch for EitherWriter
- register_rocksdb_batch() - registers batch with pending batches
- collect_shards_for_unwind() - shared logic for history shard unwinding
- Add RocksDBIntegrity enum variant for metadata
2026-01-15 10:08:31 +00:00
yongkangc
7a9e46f8f8 fix: use proper shard logic for history indices in RocksDB write_blocks_data
Add append_account_history_shard and append_storage_history_shard methods
to RocksDBBatch that properly handle shard boundaries. Update write_blocks_data
to use these methods instead of the naive approach that just wrote to u64::MAX
shard without checking existing shards.
2026-01-15 10:07:08 +00:00
yongkangc
5b6385940d fix(rocksdb): handle sentinel-only entries in consistency check
When RocksDB history tables have entries but ALL entries are sentinel
entries (highest_block_number == u64::MAX), the consistency check was
incorrectly returning Some(0) as the unwind target. This would trigger
an assertion failure during node startup.

Sentinel entries represent "open" shards that haven't been completed
yet, meaning no actual history has been indexed. This is equivalent to
the empty table case and should be treated as a first-run scenario.

Added tests to verify this edge case is handled correctly.
2026-01-15 10:03:52 +00:00
yongkangc
98b8c9aa4f fix(rocksdb): treat empty RocksDB tables as first-run scenario
Previously, when RocksDB tables were empty but MDBX had checkpoints > 0,
the consistency check would return Some(0), triggering an assertion
failure because unwinding to block 0 is considered destructive.

This is the expected state when RocksDB is enabled for the first time
alongside existing MDBX data. The fix treats empty RocksDB tables as a
first-run/migration scenario, logging a warning instead of requesting
an unwind. The pipeline will naturally populate the tables during sync.
2026-01-15 10:03:52 +00:00
yongkangc
f8a98e7bc8 refactor(provider): extract compute_history_rank helper to reduce duplication
- Add compute_history_rank() function for shared rank/select logic
- Simplify EitherReader::storage_history_info and account_history_info
- Simplify RocksTx::history_info by using the shared helper
- Import PhantomData directly instead of using std::marker::PhantomData
- Fix clippy doc_markdown warnings for RocksDB in stub module
- Make PendingRocksDBBatches pub(crate) to fix unreachable_pub warning

Reduces duplicated rank/select code across 3 locations while preserving comments.
2026-01-15 10:02:44 +00:00
yongkangc
ef6a6d8cbc fix: clippy warnings and fmt issues 2026-01-15 10:02:05 +00:00
yongkangc
dddc3be6ae fix: use PhantomData in EitherReader to capture lifetime 'a
When rocksdb feature is disabled, the RocksDB variant is compiled out,
leaving the lifetime 'a unused and causing E0392 error.

Add PhantomData<&'a ()> to StaticFile variant to ensure the lifetime
is always used regardless of feature flags.
2026-01-15 10:01:43 +00:00
yongkangc
7bd3fd9b48 refactor(provider): simplify EitherReader and encapsulate RocksDB logic
**Problem**
- EitherReader had unnecessary PhantomData markers
- RocksDB transaction setup was duplicated in historical.rs with cfg-gated blocks
- Addressed joshieDo's feedback about RocksDB logic leaking into historical provider

**Solution**
- Remove PhantomData from EitherReader enum variants (lifetime already captured by RocksDB reference)
- Add with_rocksdb_tx helper method to RocksDBProviderFactory trait
- Refactor historical.rs to use trait method instead of duplicated cfg-gated blocks

**Changes**
- Remove PhantomData from EitherReader enum and all constructors/match arms
- Add with_rocksdb_tx to RocksDBProviderFactory trait with default implementation
- Refactor account_history_lookup and storage_history_lookup to use with_rocksdb_tx helper
- Make RocksTxRefArg type alias public for trait method

**Expected Impact**
- Cleaner EitherReader API without unnecessary PhantomData
- RocksDB transaction setup encapsulated in trait method
- Reduced cfg-gated block duplication in historical.rs
- No behavioral changes, all existing tests pass (96/97)
2026-01-15 10:01:43 +00:00
yongkangc
38f0d2ad9f feat(storage): wire RocksDB into history lookups via EitherReader
This wires RocksDB into the history lookup paths:

- Adds account_history_info and storage_history_info methods to EitherReader
- Updates HistoricalStateProviderRef to use EitherReader for lookups
- Adds RocksDBProviderFactory trait bounds to provider impls
- Uses the rank/select pattern for efficient binary search in shards
2026-01-15 10:01:43 +00:00
joshie
851d8136fd add rocksdb writer to save_blocks 2026-01-15 10:01:43 +00:00
joshie
d8a600c4a8 add rocksdb writer to save_blocks 2026-01-14 21:38:23 +00:00
Sergei Shulepov
84bdf7158c feat(storage): add tracing spans (#21072)
Co-authored-by: Sergei Shulepov <pep@tempo.xyz>
2026-01-14 21:29:52 +00:00
joshieDo
c2ddcb284e Merge branch 'main' into joshie/par-save-blocks 2026-01-14 17:39:45 +00:00
joshieDo
fadc97aa03 feat: enable account changesets on save_blocks (#21012) 2026-01-14 17:39:27 +00:00
joshieDo
40ea5938fc fix 2026-01-13 17:47:12 +00:00
joshieDo
55e0f3d0db touch-ups 2026-01-13 16:48:46 +00:00
joshieDo
ad4998d473 rm Tx: Sync bound 2026-01-13 16:48:46 +00:00
joshieDo
cf89d57395 parallelize save_blocks 2026-01-13 16:48:46 +00:00
341 changed files with 5037 additions and 11133 deletions

View File

@@ -12,7 +12,7 @@ workflows:
# Check that `A` activates the features of `B`.
"propagate-feature",
# These are the features to check:
"--features=std,op,dev,asm-keccak,jemalloc,jemalloc-prof,tracy-allocator,tracy,serde-bincode-compat,serde,test-utils,arbitrary,bench,alloy-compat,min-error-logs,min-warn-logs,min-info-logs,min-debug-logs,min-trace-logs,otlp,otlp-logs,js-tracer,portable,keccak-cache-global",
"--features=std,op,dev,asm-keccak,jemalloc,jemalloc-prof,tracy-allocator,tracy,serde-bincode-compat,serde,test-utils,arbitrary,bench,alloy-compat,min-error-logs,min-warn-logs,min-info-logs,min-debug-logs,min-trace-logs,otlp,js-tracer,portable,keccak-cache-global",
# Do not try to add a new section to `[features]` of `A` only because `B` exposes that feature. There are edge-cases where this is still needed, but we can add them manually.
"--left-side-feature-missing=ignore",
# Ignore the case that `A` it outside of the workspace. Otherwise it will report errors in external dependencies that we have no influence on.

View File

@@ -15,6 +15,6 @@ permissions:
jobs:
update:
uses: tempoxyz/ci/.github/workflows/cargo-update-pr.yml@main
uses: ithacaxyz/ci/.github/workflows/cargo-update-pr.yml@main
secrets:
token: ${{ secrets.GITHUB_TOKEN }}

View File

@@ -285,7 +285,7 @@ jobs:
- run: zepter run check
deny:
uses: tempoxyz/ci/.github/workflows/deny.yml@main
uses: ithacaxyz/ci/.github/workflows/deny.yml@main
lint-success:
name: lint success

View File

@@ -249,7 +249,7 @@ Write comments that remain valuable after the PR is merged. Future readers won't
unsafe impl GlobalAlloc for LimitedAllocator { ... }
// Binary search requires sorted input. Panics on unsorted slices.
fn find_index(items: &[Item], target: &Item) -> Option<usize>
fn find_index(items: &[Item], target: &Item) -> Option
// Timeout set to 5s to match EVM block processing limits
const TRACER_TIMEOUT: Duration = Duration::from_secs(5);

1285
Cargo.lock generated

File diff suppressed because it is too large Load Diff

View File

@@ -1,5 +1,5 @@
[workspace.package]
version = "1.10.1"
version = "1.9.3"
edition = "2024"
rust-version = "1.88"
license = "MIT OR Apache-2.0"
@@ -473,22 +473,22 @@ reth-ress-protocol = { path = "crates/ress/protocol" }
reth-ress-provider = { path = "crates/ress/provider" }
# revm
revm = { version = "34.0.0", default-features = false }
revm-bytecode = { version = "8.0.0", default-features = false }
revm-database = { version = "10.0.0", default-features = false }
revm-state = { version = "9.0.0", default-features = false }
revm-primitives = { version = "22.0.0", default-features = false }
revm-interpreter = { version = "32.0.0", default-features = false }
revm-database-interface = { version = "9.0.0", default-features = false }
op-revm = { version = "15.0.0", default-features = false }
revm-inspectors = "0.34.0"
revm = { version = "33.1.0", default-features = false }
revm-bytecode = { version = "7.1.1", default-features = false }
revm-database = { version = "9.0.5", default-features = false }
revm-state = { version = "8.1.1", default-features = false }
revm-primitives = { version = "21.0.2", default-features = false }
revm-interpreter = { version = "31.1.0", default-features = false }
revm-database-interface = { version = "8.0.5", default-features = false }
op-revm = { version = "14.1.0", default-features = false }
revm-inspectors = "0.33.2"
# eth
alloy-chains = { version = "0.2.5", default-features = false }
alloy-dyn-abi = "1.4.3"
alloy-dyn-abi = "1.4.1"
alloy-eip2124 = { version = "0.2.0", default-features = false }
alloy-eip7928 = { version = "0.3.0", default-features = false }
alloy-evm = { version = "0.26.3", default-features = false }
alloy-eip7928 = { version = "0.1.0", default-features = false }
alloy-evm = { version = "0.25.1", default-features = false }
alloy-primitives = { version = "1.5.0", default-features = false, features = ["map-foldhash"] }
alloy-rlp = { version = "0.3.10", default-features = false, features = ["core-net"] }
alloy-sol-macro = "1.5.0"
@@ -497,36 +497,36 @@ alloy-trie = { version = "0.9.1", default-features = false }
alloy-hardforks = "0.4.5"
alloy-consensus = { version = "1.4.3", default-features = false }
alloy-contract = { version = "1.4.3", default-features = false }
alloy-eips = { version = "1.4.3", default-features = false }
alloy-genesis = { version = "1.4.3", default-features = false }
alloy-json-rpc = { version = "1.4.3", default-features = false }
alloy-network = { version = "1.4.3", default-features = false }
alloy-network-primitives = { version = "1.4.3", default-features = false }
alloy-provider = { version = "1.4.3", features = ["reqwest", "debug-api"], default-features = false }
alloy-pubsub = { version = "1.4.3", default-features = false }
alloy-rpc-client = { version = "1.4.3", default-features = false }
alloy-rpc-types = { version = "1.4.3", features = ["eth"], default-features = false }
alloy-rpc-types-admin = { version = "1.4.3", default-features = false }
alloy-rpc-types-anvil = { version = "1.4.3", default-features = false }
alloy-rpc-types-beacon = { version = "1.4.3", default-features = false }
alloy-rpc-types-debug = { version = "1.4.3", default-features = false }
alloy-rpc-types-engine = { version = "1.4.3", default-features = false }
alloy-rpc-types-eth = { version = "1.4.3", default-features = false }
alloy-rpc-types-mev = { version = "1.4.3", default-features = false }
alloy-rpc-types-trace = { version = "1.4.3", default-features = false }
alloy-rpc-types-txpool = { version = "1.4.3", default-features = false }
alloy-serde = { version = "1.4.3", default-features = false }
alloy-signer = { version = "1.4.3", default-features = false }
alloy-signer-local = { version = "1.4.3", default-features = false }
alloy-transport = { version = "1.4.3" }
alloy-transport-http = { version = "1.4.3", features = ["reqwest-rustls-tls"], default-features = false }
alloy-transport-ipc = { version = "1.4.3", default-features = false }
alloy-transport-ws = { version = "1.4.3", default-features = false }
alloy-consensus = { version = "1.4.1", default-features = false }
alloy-contract = { version = "1.4.1", default-features = false }
alloy-eips = { version = "1.4.1", default-features = false }
alloy-genesis = { version = "1.4.1", default-features = false }
alloy-json-rpc = { version = "1.4.1", default-features = false }
alloy-network = { version = "1.4.1", default-features = false }
alloy-network-primitives = { version = "1.4.1", default-features = false }
alloy-provider = { version = "1.4.1", features = ["reqwest", "debug-api"], default-features = false }
alloy-pubsub = { version = "1.4.1", default-features = false }
alloy-rpc-client = { version = "1.4.1", default-features = false }
alloy-rpc-types = { version = "1.4.1", features = ["eth"], default-features = false }
alloy-rpc-types-admin = { version = "1.4.1", default-features = false }
alloy-rpc-types-anvil = { version = "1.4.1", default-features = false }
alloy-rpc-types-beacon = { version = "1.4.1", default-features = false }
alloy-rpc-types-debug = { version = "1.4.1", default-features = false }
alloy-rpc-types-engine = { version = "1.4.1", default-features = false }
alloy-rpc-types-eth = { version = "1.4.1", default-features = false }
alloy-rpc-types-mev = { version = "1.4.1", default-features = false }
alloy-rpc-types-trace = { version = "1.4.1", default-features = false }
alloy-rpc-types-txpool = { version = "1.4.1", default-features = false }
alloy-serde = { version = "1.4.1", default-features = false }
alloy-signer = { version = "1.4.1", default-features = false }
alloy-signer-local = { version = "1.4.1", default-features = false }
alloy-transport = { version = "1.4.1" }
alloy-transport-http = { version = "1.4.1", features = ["reqwest-rustls-tls"], default-features = false }
alloy-transport-ipc = { version = "1.4.1", default-features = false }
alloy-transport-ws = { version = "1.4.1", default-features = false }
# op
alloy-op-evm = { version = "0.26.3", default-features = false }
alloy-op-evm = { version = "0.25.0", default-features = false }
alloy-op-hardforks = "0.4.4"
op-alloy-rpc-types = { version = "0.23.1", default-features = false }
op-alloy-rpc-types-engine = { version = "0.23.1", default-features = false }
@@ -665,7 +665,6 @@ opentelemetry_sdk = "0.31"
opentelemetry = "0.31"
opentelemetry-otlp = "0.31"
opentelemetry-semantic-conventions = "0.31"
opentelemetry-appender-tracing = "0.31"
tracing-opentelemetry = "0.32"
# misc-testing
@@ -739,15 +738,15 @@ tracing-subscriber = { version = "0.3", default-features = false }
tracing-tracy = "0.11"
triehash = "0.8"
typenum = "1.15.0"
vergen = "9.1.0"
vergen = "9.0.4"
visibility = "0.1.1"
walkdir = "2.3.3"
vergen-git2 = "9.1.0"
vergen-git2 = "1.0.5"
# networking
ipnet = "2.11"
[patch.crates-io]
# [patch.crates-io]
# alloy-consensus = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
# alloy-contract = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
# alloy-eips = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
@@ -792,8 +791,3 @@ ipnet = "2.11"
# alloy-evm = { git = "https://github.com/alloy-rs/evm", rev = "a69f0b45a6b0286e16072cb8399e02ce6ceca353" }
# alloy-op-evm = { git = "https://github.com/alloy-rs/evm", rev = "a69f0b45a6b0286e16072cb8399e02ce6ceca353" }
# revm-inspectors = { git = "https://github.com/paradigmxyz/revm-inspectors", rev = "3020ea8" }
# alloy-evm = { git = "https://github.com/alloy-rs/evm", rev = "072c248" }
# alloy-op-evm = { git = "https://github.com/alloy-rs/evm", rev = "072c248" }

View File

@@ -163,7 +163,6 @@ impl NodeManager {
"eth,reth".to_string(),
"--disable-discovery".to_string(),
"--trusted-only".to_string(),
"--disable-tx-gossip".to_string(),
]);
// Add tracing arguments if OTLP endpoint is configured

View File

@@ -17,26 +17,21 @@ workspace = true
reth-cli-runner.workspace = true
reth-cli-util.workspace = true
reth-engine-primitives.workspace = true
reth-ethereum-primitives.workspace = true
reth-fs-util.workspace = true
reth-node-api.workspace = true
reth-node-core.workspace = true
reth-primitives-traits.workspace = true
reth-rpc-api.workspace = true
reth-tracing.workspace = true
reth-chainspec.workspace = true
# alloy
alloy-eips.workspace = true
alloy-json-rpc.workspace = true
alloy-consensus.workspace = true
alloy-network.workspace = true
alloy-primitives.workspace = true
alloy-provider = { workspace = true, features = ["engine-api", "pubsub", "reqwest-rustls-tls"], default-features = false }
alloy-pubsub.workspace = true
alloy-rpc-client = { workspace = true, features = ["pubsub"] }
alloy-rpc-types-engine = { workspace = true, features = ["kzg"] }
alloy-rpc-types-engine.workspace = true
alloy-transport-http.workspace = true
alloy-transport-ipc.workspace = true
alloy-transport-ws.workspace = true

View File

@@ -1,160 +0,0 @@
//! Benchmarks empty block processing by ramping the block gas limit.
use crate::{
authenticated_transport::AuthenticatedTransportConnect,
bench::{
helpers::{build_payload, prepare_payload_request, rpc_block_to_header},
output::GasRampPayloadFile,
},
valid_payload::{call_forkchoice_updated, call_new_payload, payload_to_new_payload},
};
use alloy_eips::BlockNumberOrTag;
use alloy_provider::{network::AnyNetwork, Provider, RootProvider};
use alloy_rpc_client::ClientBuilder;
use alloy_rpc_types_engine::{ExecutionPayload, ForkchoiceState, JwtSecret};
use clap::Parser;
use reqwest::Url;
use reth_chainspec::ChainSpec;
use reth_cli_runner::CliContext;
use reth_ethereum_primitives::TransactionSigned;
use reth_primitives_traits::constants::{GAS_LIMIT_BOUND_DIVISOR, MAXIMUM_GAS_LIMIT_BLOCK};
use std::{path::PathBuf, time::Instant};
use tracing::info;
/// `reth benchmark gas-limit-ramp` command.
#[derive(Debug, Parser)]
pub struct Command {
/// Number of blocks to generate.
#[arg(long, value_name = "BLOCKS")]
blocks: u64,
/// The Engine API RPC URL.
#[arg(long = "engine-rpc-url", value_name = "ENGINE_RPC_URL")]
engine_rpc_url: String,
/// Path to the JWT secret for Engine API authentication.
#[arg(long = "jwt-secret", value_name = "JWT_SECRET")]
jwt_secret: PathBuf,
/// Output directory for benchmark results and generated payloads.
#[arg(long, value_name = "OUTPUT")]
output: PathBuf,
}
impl Command {
/// Execute `benchmark gas-limit-ramp` command.
pub async fn execute(self, _ctx: CliContext) -> eyre::Result<()> {
if self.blocks == 0 {
return Err(eyre::eyre!("--blocks must be greater than 0"));
}
// Ensure output directory exists
if self.output.is_file() {
return Err(eyre::eyre!("Output path must be a directory"));
}
if !self.output.exists() {
std::fs::create_dir_all(&self.output)?;
info!("Created output directory: {:?}", self.output);
}
// Set up authenticated provider (used for both Engine API and eth_ methods)
let jwt = std::fs::read_to_string(&self.jwt_secret)?;
let jwt = JwtSecret::from_hex(jwt)?;
let auth_url = Url::parse(&self.engine_rpc_url)?;
info!("Connecting to Engine RPC at {}", auth_url);
let auth_transport = AuthenticatedTransportConnect::new(auth_url, jwt);
let client = ClientBuilder::default().connect_with(auth_transport).await?;
let provider = RootProvider::<AnyNetwork>::new(client);
// Get chain spec - required for fork detection
let chain_id = provider.get_chain_id().await?;
let chain_spec = ChainSpec::from_chain_id(chain_id)
.ok_or_else(|| eyre::eyre!("Unsupported chain id: {chain_id}"))?;
// Fetch the current head block as parent
let parent_block = provider
.get_block_by_number(BlockNumberOrTag::Latest)
.full()
.await?
.ok_or_else(|| eyre::eyre!("Failed to fetch latest block"))?;
let (mut parent_header, mut parent_hash) = rpc_block_to_header(parent_block);
let canonical_parent = parent_header.number;
let start_block = canonical_parent + 1;
let end_block = start_block + self.blocks - 1;
info!(canonical_parent, start_block, end_block, "Starting gas limit ramp benchmark");
let mut next_block_number = start_block;
let total_benchmark_duration = Instant::now();
while next_block_number <= end_block {
let timestamp = parent_header.timestamp.saturating_add(1);
let request = prepare_payload_request(&chain_spec, timestamp, parent_hash);
let new_payload_version = request.new_payload_version;
let (payload, sidecar) = build_payload(&provider, request).await?;
let mut block =
payload.clone().try_into_block_with_sidecar::<TransactionSigned>(&sidecar)?;
let max_increase = max_gas_limit_increase(parent_header.gas_limit);
let gas_limit =
parent_header.gas_limit.saturating_add(max_increase).min(MAXIMUM_GAS_LIMIT_BLOCK);
block.header.gas_limit = gas_limit;
let block_hash = block.header.hash_slow();
// Regenerate the payload from the modified block, but keep the original sidecar
// which contains the actual execution requests data (not just the hash)
let (payload, _) = ExecutionPayload::from_block_unchecked(block_hash, &block);
let (version, params) = payload_to_new_payload(
payload,
sidecar,
false,
block.header.withdrawals_root,
Some(new_payload_version),
)?;
// Save payload to file with version info for replay
let payload_path =
self.output.join(format!("payload_block_{}.json", block.header.number));
let file =
GasRampPayloadFile { version: version as u8, block_hash, params: params.clone() };
let payload_json = serde_json::to_string_pretty(&file)?;
std::fs::write(&payload_path, &payload_json)?;
info!(block_number = block.header.number, path = %payload_path.display(), "Saved payload");
call_new_payload(&provider, version, params).await?;
let forkchoice_state = ForkchoiceState {
head_block_hash: block_hash,
safe_block_hash: block_hash,
finalized_block_hash: block_hash,
};
call_forkchoice_updated(&provider, version, forkchoice_state, None).await?;
parent_header = block.header;
parent_hash = block_hash;
next_block_number += 1;
}
let final_gas_limit = parent_header.gas_limit;
info!(
total_duration=?total_benchmark_duration.elapsed(),
blocks_processed = self.blocks,
final_gas_limit,
"Benchmark complete"
);
Ok(())
}
}
const fn max_gas_limit_increase(parent_gas_limit: u64) -> u64 {
(parent_gas_limit / GAS_LIMIT_BOUND_DIVISOR).saturating_sub(1)
}

View File

@@ -1,617 +0,0 @@
//! Command for generating large blocks by packing transactions from real blocks.
//!
//! This command fetches transactions from existing blocks and packs them into a single
//! large block using the `testing_buildBlockV1` RPC endpoint.
use crate::authenticated_transport::AuthenticatedTransportConnect;
use alloy_eips::{BlockNumberOrTag, Typed2718};
use alloy_primitives::{Bytes, B256};
use alloy_provider::{ext::EngineApi, network::AnyNetwork, Provider, RootProvider};
use alloy_rpc_client::ClientBuilder;
use alloy_rpc_types_engine::{
ExecutionPayloadEnvelopeV4, ExecutionPayloadEnvelopeV5, ForkchoiceState, JwtSecret,
PayloadAttributes,
};
use alloy_transport::layers::RetryBackoffLayer;
use clap::Parser;
use eyre::Context;
use reqwest::Url;
use reth_cli_runner::CliContext;
use reth_rpc_api::TestingBuildBlockRequestV1;
use std::future::Future;
use tokio::sync::mpsc;
use tracing::{info, warn};
/// A single transaction with its gas used and raw encoded bytes.
#[derive(Debug, Clone)]
pub struct RawTransaction {
/// The actual gas used by the transaction (from receipt).
pub gas_used: u64,
/// The transaction type (e.g., 3 for EIP-4844 blob txs).
pub tx_type: u8,
/// The raw RLP-encoded transaction bytes.
pub raw: Bytes,
}
/// Abstraction over sources of transactions for big block generation.
///
/// Implementors provide transactions from different sources (RPC, database, files, etc.)
pub trait TransactionSource {
/// Fetch transactions from a specific block number.
///
/// Returns `Ok(None)` if the block doesn't exist.
/// Returns `Ok(Some((transactions, gas_used)))` with the block's transactions and total gas.
fn fetch_block_transactions(
&self,
block_number: u64,
) -> impl Future<Output = eyre::Result<Option<(Vec<RawTransaction>, u64)>>> + Send;
}
/// RPC-based transaction source that fetches from a remote node.
#[derive(Debug)]
pub struct RpcTransactionSource {
provider: RootProvider<AnyNetwork>,
}
impl RpcTransactionSource {
/// Create a new RPC transaction source.
pub const fn new(provider: RootProvider<AnyNetwork>) -> Self {
Self { provider }
}
/// Create from an RPC URL with retry backoff.
pub fn from_url(rpc_url: &str) -> eyre::Result<Self> {
let client = ClientBuilder::default()
.layer(RetryBackoffLayer::new(10, 800, u64::MAX))
.http(rpc_url.parse()?);
let provider = RootProvider::<AnyNetwork>::new(client);
Ok(Self { provider })
}
}
impl TransactionSource for RpcTransactionSource {
async fn fetch_block_transactions(
&self,
block_number: u64,
) -> eyre::Result<Option<(Vec<RawTransaction>, u64)>> {
// Fetch block and receipts in parallel
let (block, receipts) = tokio::try_join!(
self.provider.get_block_by_number(block_number.into()).full(),
self.provider.get_block_receipts(block_number.into())
)?;
let Some(block) = block else {
return Ok(None);
};
let Some(receipts) = receipts else {
return Err(eyre::eyre!("Receipts not found for block {}", block_number));
};
let block_gas_used = block.header.gas_used;
// Convert cumulative gas from receipts to per-tx gas_used
let mut prev_cumulative = 0u64;
let transactions: Vec<RawTransaction> = block
.transactions
.txns()
.zip(receipts.iter())
.map(|(tx, receipt)| {
let cumulative = receipt.inner.inner.inner.receipt.cumulative_gas_used;
let gas_used = cumulative - prev_cumulative;
prev_cumulative = cumulative;
let with_encoded = tx.inner.inner.clone().into_encoded();
RawTransaction {
gas_used,
tx_type: tx.inner.ty(),
raw: with_encoded.encoded_bytes().clone(),
}
})
.collect();
Ok(Some((transactions, block_gas_used)))
}
}
/// Collects transactions from a source up to a target gas usage.
#[derive(Debug)]
pub struct TransactionCollector<S> {
source: S,
target_gas: u64,
}
impl<S: TransactionSource> TransactionCollector<S> {
/// Create a new transaction collector.
pub const fn new(source: S, target_gas: u64) -> Self {
Self { source, target_gas }
}
/// Collect transactions starting from the given block number.
///
/// Skips blob transactions (type 3) and collects until target gas is reached.
/// Returns the collected raw transaction bytes, total gas used, and the next block number.
pub async fn collect(&self, start_block: u64) -> eyre::Result<(Vec<Bytes>, u64, u64)> {
let mut transactions: Vec<Bytes> = Vec::new();
let mut total_gas: u64 = 0;
let mut current_block = start_block;
while total_gas < self.target_gas {
let Some((block_txs, _)) = self.source.fetch_block_transactions(current_block).await?
else {
warn!(block = current_block, "Block not found, stopping");
break;
};
for tx in block_txs {
// Skip blob transactions (EIP-4844, type 3)
if tx.tx_type == 3 {
continue;
}
if total_gas + tx.gas_used <= self.target_gas {
transactions.push(tx.raw);
total_gas += tx.gas_used;
}
if total_gas >= self.target_gas {
break;
}
}
current_block += 1;
// Stop early if remaining gas is under 1M (close enough to target)
let remaining_gas = self.target_gas.saturating_sub(total_gas);
if remaining_gas < 1_000_000 {
break;
}
}
info!(
total_txs = transactions.len(),
total_gas,
next_block = current_block,
"Finished collecting transactions"
);
Ok((transactions, total_gas, current_block))
}
}
/// `reth bench generate-big-block` command
///
/// Generates a large block by fetching transactions from existing blocks and packing them
/// into a single block using the `testing_buildBlockV1` RPC endpoint.
#[derive(Debug, Parser)]
pub struct Command {
/// The RPC URL to use for fetching blocks (can be an external archive node).
#[arg(long, value_name = "RPC_URL")]
rpc_url: String,
/// The engine RPC URL (with JWT authentication).
#[arg(long, value_name = "ENGINE_RPC_URL", default_value = "http://localhost:8551")]
engine_rpc_url: String,
/// The RPC URL for `testing_buildBlockV1` calls (same node as engine, regular RPC port).
#[arg(long, value_name = "TESTING_RPC_URL", default_value = "http://localhost:8545")]
testing_rpc_url: String,
/// Path to the JWT secret file for engine API authentication.
#[arg(long, value_name = "JWT_SECRET")]
jwt_secret: std::path::PathBuf,
/// Target gas to pack into the block.
#[arg(long, value_name = "TARGET_GAS", default_value = "30000000")]
target_gas: u64,
/// Starting block number to fetch transactions from.
/// If not specified, starts from the engine's latest block.
#[arg(long, value_name = "FROM_BLOCK")]
from_block: Option<u64>,
/// Execute the payload (call newPayload + forkchoiceUpdated).
/// If false, only builds the payload and prints it.
#[arg(long, default_value = "false")]
execute: bool,
/// Number of payloads to generate. Each payload uses the previous as parent.
/// When count == 1, the payload is only generated and saved, not executed.
/// When count > 1, each payload is executed before building the next.
#[arg(long, default_value = "1")]
count: u64,
/// Number of transaction batches to prefetch in background when count > 1.
/// Higher values reduce latency but use more memory.
#[arg(long, default_value = "4")]
prefetch_buffer: usize,
/// Output directory for generated payloads. Each payload is saved as `payload_block_N.json`.
#[arg(long, value_name = "OUTPUT_DIR")]
output_dir: std::path::PathBuf,
}
/// A built payload ready for execution.
struct BuiltPayload {
block_number: u64,
envelope: ExecutionPayloadEnvelopeV4,
block_hash: B256,
timestamp: u64,
}
impl Command {
/// Execute the `generate-big-block` command
pub async fn execute(self, _ctx: CliContext) -> eyre::Result<()> {
info!(target_gas = self.target_gas, count = self.count, "Generating big block(s)");
// Set up authenticated engine provider
let jwt =
std::fs::read_to_string(&self.jwt_secret).wrap_err("Failed to read JWT secret file")?;
let jwt = JwtSecret::from_hex(jwt.trim())?;
let auth_url = Url::parse(&self.engine_rpc_url)?;
info!("Connecting to Engine RPC at {}", auth_url);
let auth_transport = AuthenticatedTransportConnect::new(auth_url.clone(), jwt);
let auth_client = ClientBuilder::default().connect_with(auth_transport).await?;
let auth_provider = RootProvider::<AnyNetwork>::new(auth_client);
// Set up testing RPC provider (for testing_buildBlockV1)
info!("Connecting to Testing RPC at {}", self.testing_rpc_url);
let testing_client = ClientBuilder::default()
.layer(RetryBackoffLayer::new(10, 800, u64::MAX))
.http(self.testing_rpc_url.parse()?);
let testing_provider = RootProvider::<AnyNetwork>::new(testing_client);
// Get the parent block (latest canonical block)
info!(endpoint = "engine", method = "eth_getBlockByNumber", block = "latest", "RPC call");
let parent_block = auth_provider
.get_block_by_number(BlockNumberOrTag::Latest)
.await?
.ok_or_else(|| eyre::eyre!("Failed to fetch latest block"))?;
let parent_hash = parent_block.header.hash;
let parent_number = parent_block.header.number;
let parent_timestamp = parent_block.header.timestamp;
info!(
parent_hash = %parent_hash,
parent_number = parent_number,
"Using initial parent block"
);
// Create output directory
std::fs::create_dir_all(&self.output_dir).wrap_err_with(|| {
format!("Failed to create output directory: {:?}", self.output_dir)
})?;
let start_block = self.from_block.unwrap_or(parent_number);
// Use pipelined execution when generating multiple payloads
if self.count > 1 {
self.execute_pipelined(
&auth_provider,
&testing_provider,
start_block,
parent_hash,
parent_timestamp,
)
.await?;
} else {
// Single payload - collect transactions and build
let tx_source = RpcTransactionSource::from_url(&self.rpc_url)?;
let collector = TransactionCollector::new(tx_source, self.target_gas);
let (transactions, _total_gas, _next_block) = collector.collect(start_block).await?;
if transactions.is_empty() {
return Err(eyre::eyre!("No transactions collected"));
}
self.execute_sequential(
&auth_provider,
&testing_provider,
transactions,
parent_hash,
parent_timestamp,
)
.await?;
}
info!(count = self.count, output_dir = %self.output_dir.display(), "All payloads generated");
Ok(())
}
/// Sequential execution path for single payload or no-execute mode.
async fn execute_sequential(
&self,
auth_provider: &RootProvider<AnyNetwork>,
testing_provider: &RootProvider<AnyNetwork>,
transactions: Vec<Bytes>,
mut parent_hash: B256,
mut parent_timestamp: u64,
) -> eyre::Result<()> {
for i in 0..self.count {
info!(
payload = i + 1,
total = self.count,
parent_hash = %parent_hash,
parent_timestamp = parent_timestamp,
"Building payload via testing_buildBlockV1"
);
let built = self
.build_payload(testing_provider, &transactions, i, parent_hash, parent_timestamp)
.await?;
self.save_payload(&built)?;
if self.execute || self.count > 1 {
info!(payload = i + 1, block_hash = %built.block_hash, "Executing payload (newPayload + FCU)");
self.execute_payload_v4(auth_provider, built.envelope, parent_hash).await?;
info!(payload = i + 1, "Payload executed successfully");
}
parent_hash = built.block_hash;
parent_timestamp = built.timestamp;
}
Ok(())
}
/// Pipelined execution - fetches transactions and builds payloads in background.
async fn execute_pipelined(
&self,
auth_provider: &RootProvider<AnyNetwork>,
testing_provider: &RootProvider<AnyNetwork>,
start_block: u64,
initial_parent_hash: B256,
initial_parent_timestamp: u64,
) -> eyre::Result<()> {
// Create channel for transaction batches (one batch per payload)
let (tx_sender, mut tx_receiver) = mpsc::channel::<Vec<Bytes>>(self.prefetch_buffer);
// Spawn background task to continuously fetch transaction batches
let rpc_url = self.rpc_url.clone();
let target_gas = self.target_gas;
let count = self.count;
let fetcher_handle = tokio::spawn(async move {
let tx_source = match RpcTransactionSource::from_url(&rpc_url) {
Ok(source) => source,
Err(e) => {
warn!(error = %e, "Failed to create transaction source");
return;
}
};
let collector = TransactionCollector::new(tx_source, target_gas);
let mut current_block = start_block;
for payload_idx in 0..count {
match collector.collect(current_block).await {
Ok((transactions, total_gas, next_block)) => {
info!(
payload = payload_idx + 1,
tx_count = transactions.len(),
total_gas,
blocks = format!("{}..{}", current_block, next_block),
"Fetched transactions"
);
current_block = next_block;
if tx_sender.send(transactions).await.is_err() {
break;
}
}
Err(e) => {
warn!(payload = payload_idx + 1, error = %e, "Failed to fetch transactions");
break;
}
}
}
});
let mut parent_hash = initial_parent_hash;
let mut parent_timestamp = initial_parent_timestamp;
let mut pending_build: Option<tokio::task::JoinHandle<eyre::Result<BuiltPayload>>> = None;
for i in 0..self.count {
let is_last = i == self.count - 1;
// Get current payload (either from pending build or build now)
let current_payload = if let Some(handle) = pending_build.take() {
handle.await??
} else {
// First payload - wait for transactions and build synchronously
let transactions = tx_receiver
.recv()
.await
.ok_or_else(|| eyre::eyre!("Transaction fetcher stopped unexpectedly"))?;
if transactions.is_empty() {
return Err(eyre::eyre!("No transactions collected for payload {}", i + 1));
}
info!(
payload = i + 1,
total = self.count,
parent_hash = %parent_hash,
parent_timestamp = parent_timestamp,
tx_count = transactions.len(),
"Building payload via testing_buildBlockV1"
);
self.build_payload(
testing_provider,
&transactions,
i,
parent_hash,
parent_timestamp,
)
.await?
};
self.save_payload(&current_payload)?;
let current_block_hash = current_payload.block_hash;
let current_timestamp = current_payload.timestamp;
// Execute current payload first
info!(payload = i + 1, block_hash = %current_block_hash, "Executing payload (newPayload + FCU)");
self.execute_payload_v4(auth_provider, current_payload.envelope, parent_hash).await?;
info!(payload = i + 1, "Payload executed successfully");
// Start building next payload in background (if not last) - AFTER execution
if !is_last {
// Get transactions for next payload (should already be fetched or fetching)
let next_transactions = tx_receiver
.recv()
.await
.ok_or_else(|| eyre::eyre!("Transaction fetcher stopped unexpectedly"))?;
if next_transactions.is_empty() {
return Err(eyre::eyre!("No transactions collected for payload {}", i + 2));
}
let testing_provider = testing_provider.clone();
let next_index = i + 1;
let total = self.count;
pending_build = Some(tokio::spawn(async move {
info!(
payload = next_index + 1,
total = total,
parent_hash = %current_block_hash,
parent_timestamp = current_timestamp,
tx_count = next_transactions.len(),
"Building payload via testing_buildBlockV1"
);
Self::build_payload_static(
&testing_provider,
&next_transactions,
next_index,
current_block_hash,
current_timestamp,
)
.await
}));
}
parent_hash = current_block_hash;
parent_timestamp = current_timestamp;
}
// Clean up the fetcher task
drop(tx_receiver);
let _ = fetcher_handle.await;
Ok(())
}
/// Build a single payload via `testing_buildBlockV1`.
async fn build_payload(
&self,
testing_provider: &RootProvider<AnyNetwork>,
transactions: &[Bytes],
index: u64,
parent_hash: B256,
parent_timestamp: u64,
) -> eyre::Result<BuiltPayload> {
Self::build_payload_static(
testing_provider,
transactions,
index,
parent_hash,
parent_timestamp,
)
.await
}
/// Static version for use in spawned tasks.
async fn build_payload_static(
testing_provider: &RootProvider<AnyNetwork>,
transactions: &[Bytes],
index: u64,
parent_hash: B256,
parent_timestamp: u64,
) -> eyre::Result<BuiltPayload> {
let request = TestingBuildBlockRequestV1 {
parent_block_hash: parent_hash,
payload_attributes: PayloadAttributes {
timestamp: parent_timestamp + 12,
prev_randao: B256::ZERO,
suggested_fee_recipient: alloy_primitives::Address::ZERO,
withdrawals: Some(vec![]),
parent_beacon_block_root: Some(B256::ZERO),
},
transactions: transactions.to_vec(),
extra_data: None,
};
let total_tx_bytes: usize = transactions.iter().map(|tx| tx.len()).sum();
info!(
payload = index + 1,
tx_count = transactions.len(),
total_tx_bytes = total_tx_bytes,
parent_hash = %parent_hash,
"Sending to testing_buildBlockV1"
);
let envelope: ExecutionPayloadEnvelopeV5 =
testing_provider.client().request("testing_buildBlockV1", [request]).await?;
let v4_envelope = envelope.try_into_v4()?;
let inner = &v4_envelope.envelope_inner.execution_payload.payload_inner.payload_inner;
let block_hash = inner.block_hash;
let block_number = inner.block_number;
let timestamp = inner.timestamp;
Ok(BuiltPayload { block_number, envelope: v4_envelope, block_hash, timestamp })
}
/// Save a payload to disk.
fn save_payload(&self, payload: &BuiltPayload) -> eyre::Result<()> {
let filename = format!("payload_block_{}.json", payload.block_number);
let filepath = self.output_dir.join(&filename);
let json = serde_json::to_string_pretty(&payload.envelope)?;
std::fs::write(&filepath, &json)
.wrap_err_with(|| format!("Failed to write payload to {:?}", filepath))?;
info!(block_number = payload.block_number, block_hash = %payload.block_hash, path = %filepath.display(), "Payload saved");
Ok(())
}
async fn execute_payload_v4(
&self,
provider: &RootProvider<AnyNetwork>,
envelope: ExecutionPayloadEnvelopeV4,
parent_hash: B256,
) -> eyre::Result<()> {
let block_hash =
envelope.envelope_inner.execution_payload.payload_inner.payload_inner.block_hash;
let status = provider
.new_payload_v4(
envelope.envelope_inner.execution_payload,
vec![],
B256::ZERO,
envelope.execution_requests.to_vec(),
)
.await?;
if !status.is_valid() {
return Err(eyre::eyre!("Payload rejected: {:?}", status));
}
let fcu_state = ForkchoiceState {
head_block_hash: block_hash,
safe_block_hash: parent_hash,
finalized_block_hash: parent_hash,
};
let fcu_result = provider.fork_choice_updated_v3(fcu_state, None).await?;
if !fcu_result.is_valid() {
return Err(eyre::eyre!("FCU rejected: {:?}", fcu_result));
}
Ok(())
}
}

View File

@@ -1,196 +0,0 @@
//! Common helpers for reth-bench commands.
use crate::valid_payload::call_forkchoice_updated;
use alloy_consensus::Header;
use alloy_eips::eip4844::kzg_to_versioned_hash;
use alloy_primitives::{Address, B256};
use alloy_provider::{ext::EngineApi, network::AnyNetwork, RootProvider};
use alloy_rpc_types_engine::{
CancunPayloadFields, ExecutionPayload, ExecutionPayloadSidecar, ForkchoiceState,
PayloadAttributes, PayloadId, PraguePayloadFields,
};
use eyre::OptionExt;
use reth_chainspec::{ChainSpec, EthereumHardforks};
use reth_node_api::EngineApiMessageVersion;
use tracing::debug;
/// Prepared payload request data for triggering block building.
pub(crate) struct PayloadRequest {
/// The payload attributes for the new block.
pub(crate) attributes: PayloadAttributes,
/// The forkchoice state pointing to the parent block.
pub(crate) forkchoice_state: ForkchoiceState,
/// The engine API version for FCU calls.
pub(crate) fcu_version: EngineApiMessageVersion,
/// The getPayload version to use (1-5).
pub(crate) get_payload_version: u8,
/// The newPayload version to use.
pub(crate) new_payload_version: EngineApiMessageVersion,
}
/// Prepare payload attributes and forkchoice state for a new block.
pub(crate) fn prepare_payload_request(
chain_spec: &ChainSpec,
timestamp: u64,
parent_hash: B256,
) -> PayloadRequest {
let shanghai_active = chain_spec.is_shanghai_active_at_timestamp(timestamp);
let cancun_active = chain_spec.is_cancun_active_at_timestamp(timestamp);
let prague_active = chain_spec.is_prague_active_at_timestamp(timestamp);
let osaka_active = chain_spec.is_osaka_active_at_timestamp(timestamp);
// FCU version: V3 for Cancun+Prague+Osaka, V2 for Shanghai, V1 otherwise
let fcu_version = if cancun_active {
EngineApiMessageVersion::V3
} else if shanghai_active {
EngineApiMessageVersion::V2
} else {
EngineApiMessageVersion::V1
};
// getPayload version: 5 for Osaka, 4 for Prague, 3 for Cancun, 2 for Shanghai, 1 otherwise
// newPayload version: 4 for Prague+Osaka (no V5), 3 for Cancun, 2 for Shanghai, 1 otherwise
let (get_payload_version, new_payload_version) = if osaka_active {
(5, EngineApiMessageVersion::V4) // Osaka uses getPayloadV5 but newPayloadV4
} else if prague_active {
(4, EngineApiMessageVersion::V4)
} else if cancun_active {
(3, EngineApiMessageVersion::V3)
} else if shanghai_active {
(2, EngineApiMessageVersion::V2)
} else {
(1, EngineApiMessageVersion::V1)
};
PayloadRequest {
attributes: PayloadAttributes {
timestamp,
prev_randao: B256::ZERO,
suggested_fee_recipient: Address::ZERO,
withdrawals: shanghai_active.then(Vec::new),
parent_beacon_block_root: cancun_active.then_some(B256::ZERO),
},
forkchoice_state: ForkchoiceState {
head_block_hash: parent_hash,
safe_block_hash: parent_hash,
finalized_block_hash: parent_hash,
},
fcu_version,
get_payload_version,
new_payload_version,
}
}
/// Trigger payload building via FCU and retrieve the built payload.
///
/// This sends a forkchoiceUpdated with payload attributes to start building,
/// then calls getPayload to retrieve the result.
pub(crate) async fn build_payload(
provider: &RootProvider<AnyNetwork>,
request: PayloadRequest,
) -> eyre::Result<(ExecutionPayload, ExecutionPayloadSidecar)> {
let fcu_result = call_forkchoice_updated(
provider,
request.fcu_version,
request.forkchoice_state,
Some(request.attributes.clone()),
)
.await?;
let payload_id =
fcu_result.payload_id.ok_or_eyre("Payload builder did not return a payload id")?;
get_payload_with_sidecar(
provider,
request.get_payload_version,
payload_id,
request.attributes.parent_beacon_block_root,
)
.await
}
/// Convert an RPC block to a consensus header and block hash.
pub(crate) fn rpc_block_to_header(block: alloy_provider::network::AnyRpcBlock) -> (Header, B256) {
let block_hash = block.header.hash;
let header = block.header.inner.clone().into_header_with_defaults();
(header, block_hash)
}
/// Compute versioned hashes from KZG commitments.
fn versioned_hashes_from_commitments(
commitments: &[alloy_primitives::FixedBytes<48>],
) -> Vec<B256> {
commitments.iter().map(|c| kzg_to_versioned_hash(c.as_ref())).collect()
}
/// Fetch an execution payload using the appropriate engine API version.
pub(crate) async fn get_payload_with_sidecar(
provider: &RootProvider<AnyNetwork>,
version: u8,
payload_id: PayloadId,
parent_beacon_block_root: Option<B256>,
) -> eyre::Result<(ExecutionPayload, ExecutionPayloadSidecar)> {
debug!(get_payload_version = ?version, ?payload_id, "Sending getPayload");
match version {
1 => {
let payload = provider.get_payload_v1(payload_id).await?;
Ok((ExecutionPayload::V1(payload), ExecutionPayloadSidecar::none()))
}
2 => {
let envelope = provider.get_payload_v2(payload_id).await?;
let payload = match envelope.execution_payload {
alloy_rpc_types_engine::ExecutionPayloadFieldV2::V1(p) => ExecutionPayload::V1(p),
alloy_rpc_types_engine::ExecutionPayloadFieldV2::V2(p) => ExecutionPayload::V2(p),
};
Ok((payload, ExecutionPayloadSidecar::none()))
}
3 => {
let envelope = provider.get_payload_v3(payload_id).await?;
let versioned_hashes =
versioned_hashes_from_commitments(&envelope.blobs_bundle.commitments);
let cancun_fields = CancunPayloadFields {
parent_beacon_block_root: parent_beacon_block_root
.ok_or_eyre("parent_beacon_block_root required for V3")?,
versioned_hashes,
};
Ok((
ExecutionPayload::V3(envelope.execution_payload),
ExecutionPayloadSidecar::v3(cancun_fields),
))
}
4 => {
let envelope = provider.get_payload_v4(payload_id).await?;
let versioned_hashes = versioned_hashes_from_commitments(
&envelope.envelope_inner.blobs_bundle.commitments,
);
let cancun_fields = CancunPayloadFields {
parent_beacon_block_root: parent_beacon_block_root
.ok_or_eyre("parent_beacon_block_root required for V4")?,
versioned_hashes,
};
let prague_fields = PraguePayloadFields::new(envelope.execution_requests);
Ok((
ExecutionPayload::V3(envelope.envelope_inner.execution_payload),
ExecutionPayloadSidecar::v4(cancun_fields, prague_fields),
))
}
5 => {
// V5 (Osaka) - use raw request since alloy doesn't have get_payload_v5 yet
let envelope = provider.get_payload_v5(payload_id).await?;
let versioned_hashes =
versioned_hashes_from_commitments(&envelope.blobs_bundle.commitments);
let cancun_fields = CancunPayloadFields {
parent_beacon_block_root: parent_beacon_block_root
.ok_or_eyre("parent_beacon_block_root required for V5")?,
versioned_hashes,
};
let prague_fields = PraguePayloadFields::new(envelope.execution_requests);
Ok((
ExecutionPayload::V3(envelope.execution_payload),
ExecutionPayloadSidecar::v4(cancun_fields, prague_fields),
))
}
_ => panic!("This tool does not support getPayload versions past v5"),
}
}

View File

@@ -6,16 +6,9 @@ use reth_node_core::args::LogArgs;
use reth_tracing::FileWorkerGuard;
mod context;
mod gas_limit_ramp;
mod generate_big_block;
pub(crate) mod helpers;
pub use generate_big_block::{
RawTransaction, RpcTransactionSource, TransactionCollector, TransactionSource,
};
mod new_payload_fcu;
mod new_payload_only;
mod output;
mod replay_payloads;
mod send_payload;
/// `reth bench` command
@@ -34,9 +27,6 @@ pub enum Subcommands {
/// Benchmark which calls `newPayload`, then `forkchoiceUpdated`.
NewPayloadFcu(new_payload_fcu::Command),
/// Benchmark which builds empty blocks with a ramped gas limit.
GasLimitRamp(gas_limit_ramp::Command),
/// Benchmark which only calls subsequent `newPayload` calls.
NewPayloadOnly(new_payload_only::Command),
@@ -51,29 +41,6 @@ pub enum Subcommands {
/// `cast block latest --full --json | reth-bench send-payload --rpc-url localhost:5000
/// --jwt-secret $(cat ~/.local/share/reth/mainnet/jwt.hex)`
SendPayload(send_payload::Command),
/// Generate a large block by packing transactions from existing blocks.
///
/// This command fetches transactions from real blocks and packs them into a single
/// block using the `testing_buildBlockV1` RPC endpoint.
///
/// Example:
///
/// `reth-bench generate-big-block --rpc-url http://localhost:8545 --engine-rpc-url
/// http://localhost:8551 --jwt-secret ~/.local/share/reth/mainnet/jwt.hex --target-gas
/// 30000000`
GenerateBigBlock(generate_big_block::Command),
/// Replay pre-generated payloads from a directory.
///
/// This command reads payload files from a previous `generate-big-block` run and replays
/// them in sequence using `newPayload` followed by `forkchoiceUpdated`.
///
/// Example:
///
/// `reth-bench replay-payloads --payload-dir ./payloads --engine-rpc-url
/// http://localhost:8551 --jwt-secret ~/.local/share/reth/mainnet/jwt.hex`
ReplayPayloads(replay_payloads::Command),
}
impl BenchmarkCommand {
@@ -84,11 +51,8 @@ impl BenchmarkCommand {
match self.command {
Subcommands::NewPayloadFcu(command) => command.execute(ctx).await,
Subcommands::GasLimitRamp(command) => command.execute(ctx).await,
Subcommands::NewPayloadOnly(command) => command.execute(ctx).await,
Subcommands::SendPayload(command) => command.execute(ctx).await,
Subcommands::GenerateBigBlock(command) => command.execute(ctx).await,
Subcommands::ReplayPayloads(command) => command.execute(ctx).await,
}
}

View File

@@ -13,7 +13,8 @@ use crate::{
bench::{
context::BenchContext,
output::{
write_benchmark_results, CombinedResult, NewPayloadResult, TotalGasOutput, TotalGasRow,
CombinedResult, NewPayloadResult, TotalGasOutput, TotalGasRow, COMBINED_OUTPUT_SUFFIX,
GAS_OUTPUT_SUFFIX,
},
},
valid_payload::{block_to_new_payload, call_forkchoice_updated, call_new_payload},
@@ -26,6 +27,7 @@ use alloy_rpc_client::RpcClient;
use alloy_rpc_types_engine::ForkchoiceState;
use alloy_transport_ws::WsConnect;
use clap::Parser;
use csv::Writer;
use eyre::{Context, OptionExt};
use futures::StreamExt;
use humantime::parse_duration;
@@ -121,7 +123,6 @@ impl Command {
auth_provider,
mut next_block,
is_optimism,
..
} = BenchContext::new(&self.benchmark, self.rpc_url).await?;
let buffer_size = self.rpc_block_buffer_size;
@@ -187,7 +188,6 @@ impl Command {
result
} {
let gas_used = block.header.gas_used;
let gas_limit = block.header.gas_limit;
let block_number = block.header.number;
let transaction_count = block.transactions.len() as u64;
@@ -211,7 +211,6 @@ impl Command {
let fcu_latency = total_latency - new_payload_result.latency;
let combined_result = CombinedResult {
block_number,
gas_limit,
transaction_count,
new_payload_result,
fcu_latency,
@@ -241,11 +240,28 @@ impl Command {
// since the benchmark goal is measuring Ggas/s of newPayload/FCU, not persistence.
drop(waiter);
let (gas_output_results, combined_results): (Vec<TotalGasRow>, Vec<CombinedResult>) =
let (gas_output_results, combined_results): (_, Vec<CombinedResult>) =
results.into_iter().unzip();
// Write CSV output files
if let Some(ref path) = self.benchmark.output {
write_benchmark_results(path, &gas_output_results, combined_results)?;
let output_path = path.join(COMBINED_OUTPUT_SUFFIX);
info!("Writing engine api call latency output to file: {:?}", output_path);
let mut writer = Writer::from_path(&output_path)?;
for result in combined_results {
writer.serialize(result)?;
}
writer.flush()?;
let output_path = path.join(GAS_OUTPUT_SUFFIX);
info!("Writing total gas output to file: {:?}", output_path);
let mut writer = Writer::from_path(&output_path)?;
for row in &gas_output_results {
writer.serialize(row)?;
}
writer.flush()?;
info!("Finished writing benchmark output files to {:?}.", path);
}
let gas_output = TotalGasOutput::new(gas_output_results)?;

View File

@@ -49,7 +49,6 @@ impl Command {
auth_provider,
mut next_block,
is_optimism,
..
} = BenchContext::new(&self.benchmark, self.rpc_url).await?;
let buffer_size = self.rpc_block_buffer_size;
@@ -97,7 +96,11 @@ impl Command {
let transaction_count = block.transactions.len() as u64;
let gas_used = block.header.gas_used;
debug!(number=?block.header.number, "Sending payload to engine");
debug!(
target: "reth-bench",
number=?block.header.number,
"Sending payload to engine",
);
let (version, params) = block_to_new_payload(block, is_optimism)?;

View File

@@ -1,13 +1,10 @@
//! Contains various benchmark output formats, either for logging or for
//! serialization to / from files.
use alloy_primitives::B256;
use csv::Writer;
use eyre::OptionExt;
use reth_primitives_traits::constants::GIGAGAS;
use serde::{ser::SerializeStruct, Deserialize, Serialize};
use std::{path::Path, time::Duration};
use tracing::info;
use serde::{ser::SerializeStruct, Serialize};
use std::time::Duration;
/// This is the suffix for gas output csv files.
pub(crate) const GAS_OUTPUT_SUFFIX: &str = "total_gas.csv";
@@ -18,17 +15,6 @@ pub(crate) const COMBINED_OUTPUT_SUFFIX: &str = "combined_latency.csv";
/// This is the suffix for new payload output csv files.
pub(crate) const NEW_PAYLOAD_OUTPUT_SUFFIX: &str = "new_payload_latency.csv";
/// Serialized format for gas ramp payloads on disk.
#[derive(Debug, Serialize, Deserialize)]
pub(crate) struct GasRampPayloadFile {
/// Engine API version (1-5).
pub(crate) version: u8,
/// The block hash for FCU.
pub(crate) block_hash: B256,
/// The params to pass to newPayload.
pub(crate) params: serde_json::Value,
}
/// This represents the results of a single `newPayload` call in the benchmark, containing the gas
/// used and the `newPayload` latency.
#[derive(Debug)]
@@ -81,8 +67,6 @@ impl Serialize for NewPayloadResult {
pub(crate) struct CombinedResult {
/// The block number of the block being processed.
pub(crate) block_number: u64,
/// The gas limit of the block.
pub(crate) gas_limit: u64,
/// The number of transactions in the block.
pub(crate) transaction_count: u64,
/// The `newPayload` result.
@@ -104,7 +88,7 @@ impl std::fmt::Display for CombinedResult {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"Block {} processed at {:.4} Ggas/s, used {} total gas. Combined: {:.4} Ggas/s. fcu: {:?}, newPayload: {:?}",
"Payload {} processed at {:.4} Ggas/s, used {} total gas. Combined gas per second: {:.4} Ggas/s. fcu latency: {:?}, newPayload latency: {:?}",
self.block_number,
self.new_payload_result.gas_per_second() / GIGAGAS as f64,
self.new_payload_result.gas_used,
@@ -126,11 +110,10 @@ impl Serialize for CombinedResult {
let fcu_latency = self.fcu_latency.as_micros();
let new_payload_latency = self.new_payload_result.latency.as_micros();
let total_latency = self.total_latency.as_micros();
let mut state = serializer.serialize_struct("CombinedResult", 7)?;
let mut state = serializer.serialize_struct("CombinedResult", 6)?;
// flatten the new payload result because this is meant for CSV writing
state.serialize_field("block_number", &self.block_number)?;
state.serialize_field("gas_limit", &self.gas_limit)?;
state.serialize_field("transaction_count", &self.transaction_count)?;
state.serialize_field("gas_used", &self.new_payload_result.gas_used)?;
state.serialize_field("new_payload_latency", &new_payload_latency)?;
@@ -184,36 +167,6 @@ impl TotalGasOutput {
}
}
/// Write benchmark results to CSV files.
///
/// Writes two files to the output directory:
/// - `combined_latency.csv`: Per-block latency results
/// - `total_gas.csv`: Per-block gas usage over time
pub(crate) fn write_benchmark_results(
output_dir: &Path,
gas_results: &[TotalGasRow],
combined_results: Vec<CombinedResult>,
) -> eyre::Result<()> {
let output_path = output_dir.join(COMBINED_OUTPUT_SUFFIX);
info!("Writing engine api call latency output to file: {:?}", output_path);
let mut writer = Writer::from_path(&output_path)?;
for result in combined_results {
writer.serialize(result)?;
}
writer.flush()?;
let output_path = output_dir.join(GAS_OUTPUT_SUFFIX);
info!("Writing total gas output to file: {:?}", output_path);
let mut writer = Writer::from_path(&output_path)?;
for row in gas_results {
writer.serialize(row)?;
}
writer.flush()?;
info!("Finished writing benchmark output files to {:?}.", output_dir);
Ok(())
}
/// This serializes the `time` field of the [`TotalGasRow`] to microseconds.
///
/// This is essentially just for the csv writer, which would have headers

View File

@@ -1,332 +0,0 @@
//! Command for replaying pre-generated payloads from disk.
//!
//! This command reads `ExecutionPayloadEnvelopeV4` files from a directory and replays them
//! in sequence using `newPayload` followed by `forkchoiceUpdated`.
use crate::{
authenticated_transport::AuthenticatedTransportConnect,
bench::output::GasRampPayloadFile,
valid_payload::{call_forkchoice_updated, call_new_payload},
};
use alloy_primitives::B256;
use alloy_provider::{ext::EngineApi, network::AnyNetwork, Provider, RootProvider};
use alloy_rpc_client::ClientBuilder;
use alloy_rpc_types_engine::{ExecutionPayloadEnvelopeV4, ForkchoiceState, JwtSecret};
use clap::Parser;
use eyre::Context;
use reqwest::Url;
use reth_cli_runner::CliContext;
use reth_node_api::EngineApiMessageVersion;
use std::path::PathBuf;
use tracing::{debug, info};
/// `reth bench replay-payloads` command
///
/// Replays pre-generated payloads from a directory by calling `newPayload` followed by
/// `forkchoiceUpdated` for each payload in sequence.
#[derive(Debug, Parser)]
pub struct Command {
/// The engine RPC URL (with JWT authentication).
#[arg(long, value_name = "ENGINE_RPC_URL", default_value = "http://localhost:8551")]
engine_rpc_url: String,
/// Path to the JWT secret file for engine API authentication.
#[arg(long, value_name = "JWT_SECRET")]
jwt_secret: PathBuf,
/// Directory containing payload files (`payload_block_N.json`).
#[arg(long, value_name = "PAYLOAD_DIR")]
payload_dir: PathBuf,
/// Optional limit on the number of payloads to replay.
/// If not specified, replays all payloads in the directory.
#[arg(long, value_name = "COUNT")]
count: Option<usize>,
/// Skip the first N payloads.
#[arg(long, value_name = "SKIP", default_value = "0")]
skip: usize,
/// Optional directory containing gas ramp payloads to replay first.
/// These are replayed before the main payloads to warm up the gas limit.
#[arg(long, value_name = "GAS_RAMP_DIR")]
gas_ramp_dir: Option<PathBuf>,
}
/// A loaded payload ready for execution.
struct LoadedPayload {
/// The index (from filename).
index: u64,
/// The payload envelope.
envelope: ExecutionPayloadEnvelopeV4,
/// The block hash.
block_hash: B256,
}
/// A gas ramp payload loaded from disk.
struct GasRampPayload {
/// Block number from filename.
block_number: u64,
/// Engine API version for newPayload.
version: EngineApiMessageVersion,
/// The file contents.
file: GasRampPayloadFile,
}
impl Command {
/// Execute the `replay-payloads` command.
pub async fn execute(self, _ctx: CliContext) -> eyre::Result<()> {
info!(payload_dir = %self.payload_dir.display(), "Replaying payloads");
// Set up authenticated engine provider
let jwt =
std::fs::read_to_string(&self.jwt_secret).wrap_err("Failed to read JWT secret file")?;
let jwt = JwtSecret::from_hex(jwt.trim())?;
let auth_url = Url::parse(&self.engine_rpc_url)?;
info!("Connecting to Engine RPC at {}", auth_url);
let auth_transport = AuthenticatedTransportConnect::new(auth_url.clone(), jwt);
let auth_client = ClientBuilder::default().connect_with(auth_transport).await?;
let auth_provider = RootProvider::<AnyNetwork>::new(auth_client);
// Get parent block (latest canonical block) - we need this for the first FCU
let parent_block = auth_provider
.get_block_by_number(alloy_eips::BlockNumberOrTag::Latest)
.await?
.ok_or_else(|| eyre::eyre!("Failed to fetch latest block"))?;
let initial_parent_hash = parent_block.header.hash;
let initial_parent_number = parent_block.header.number;
info!(
parent_hash = %initial_parent_hash,
parent_number = initial_parent_number,
"Using initial parent block"
);
// Load all payloads upfront to avoid I/O delays between phases
let gas_ramp_payloads = if let Some(ref gas_ramp_dir) = self.gas_ramp_dir {
let payloads = self.load_gas_ramp_payloads(gas_ramp_dir)?;
if payloads.is_empty() {
return Err(eyre::eyre!("No gas ramp payload files found in {:?}", gas_ramp_dir));
}
info!(count = payloads.len(), "Loaded gas ramp payloads from disk");
payloads
} else {
Vec::new()
};
let payloads = self.load_payloads()?;
if payloads.is_empty() {
return Err(eyre::eyre!("No payload files found in {:?}", self.payload_dir));
}
info!(count = payloads.len(), "Loaded main payloads from disk");
let mut parent_hash = initial_parent_hash;
// Replay gas ramp payloads first
for (i, payload) in gas_ramp_payloads.iter().enumerate() {
info!(
gas_ramp_payload = i + 1,
total = gas_ramp_payloads.len(),
block_number = payload.block_number,
block_hash = %payload.file.block_hash,
"Executing gas ramp payload (newPayload + FCU)"
);
call_new_payload(&auth_provider, payload.version, payload.file.params.clone()).await?;
let fcu_state = ForkchoiceState {
head_block_hash: payload.file.block_hash,
safe_block_hash: parent_hash,
finalized_block_hash: parent_hash,
};
call_forkchoice_updated(&auth_provider, payload.version, fcu_state, None).await?;
info!(gas_ramp_payload = i + 1, "Gas ramp payload executed successfully");
parent_hash = payload.file.block_hash;
}
if !gas_ramp_payloads.is_empty() {
info!(count = gas_ramp_payloads.len(), "All gas ramp payloads replayed");
}
for (i, payload) in payloads.iter().enumerate() {
info!(
payload = i + 1,
total = payloads.len(),
index = payload.index,
block_hash = %payload.block_hash,
"Executing payload (newPayload + FCU)"
);
self.execute_payload_v4(&auth_provider, &payload.envelope, parent_hash).await?;
info!(payload = i + 1, "Payload executed successfully");
parent_hash = payload.block_hash;
}
info!(count = payloads.len(), "All payloads replayed successfully");
Ok(())
}
/// Load and parse all payload files from the directory.
fn load_payloads(&self) -> eyre::Result<Vec<LoadedPayload>> {
let mut payloads = Vec::new();
// Read directory entries
let entries: Vec<_> = std::fs::read_dir(&self.payload_dir)
.wrap_err_with(|| format!("Failed to read directory {:?}", self.payload_dir))?
.filter_map(|e| e.ok())
.filter(|e| {
e.path().extension().and_then(|s| s.to_str()) == Some("json") &&
e.file_name().to_string_lossy().starts_with("payload_")
})
.collect();
// Parse filenames to get indices and sort
let mut indexed_paths: Vec<(u64, PathBuf)> = entries
.into_iter()
.filter_map(|e| {
let name = e.file_name();
let name_str = name.to_string_lossy();
// Extract index from "payload_NNN.json"
let index_str = name_str.strip_prefix("payload_")?.strip_suffix(".json")?;
let index: u64 = index_str.parse().ok()?;
Some((index, e.path()))
})
.collect();
indexed_paths.sort_by_key(|(idx, _)| *idx);
// Apply skip and count
let indexed_paths: Vec<_> = indexed_paths.into_iter().skip(self.skip).collect();
let indexed_paths: Vec<_> = match self.count {
Some(count) => indexed_paths.into_iter().take(count).collect(),
None => indexed_paths,
};
// Load each payload
for (index, path) in indexed_paths {
let content = std::fs::read_to_string(&path)
.wrap_err_with(|| format!("Failed to read {:?}", path))?;
let envelope: ExecutionPayloadEnvelopeV4 = serde_json::from_str(&content)
.wrap_err_with(|| format!("Failed to parse {:?}", path))?;
let block_hash =
envelope.envelope_inner.execution_payload.payload_inner.payload_inner.block_hash;
info!(
index = index,
block_hash = %block_hash,
path = %path.display(),
"Loaded payload"
);
payloads.push(LoadedPayload { index, envelope, block_hash });
}
Ok(payloads)
}
/// Load and parse gas ramp payload files from a directory.
fn load_gas_ramp_payloads(&self, dir: &PathBuf) -> eyre::Result<Vec<GasRampPayload>> {
let mut payloads = Vec::new();
let entries: Vec<_> = std::fs::read_dir(dir)
.wrap_err_with(|| format!("Failed to read directory {:?}", dir))?
.filter_map(|e| e.ok())
.filter(|e| {
e.path().extension().and_then(|s| s.to_str()) == Some("json") &&
e.file_name().to_string_lossy().starts_with("payload_block_")
})
.collect();
// Parse filenames to get block numbers and sort
let mut indexed_paths: Vec<(u64, PathBuf)> = entries
.into_iter()
.filter_map(|e| {
let name = e.file_name();
let name_str = name.to_string_lossy();
// Extract block number from "payload_block_NNN.json"
let block_str = name_str.strip_prefix("payload_block_")?.strip_suffix(".json")?;
let block_number: u64 = block_str.parse().ok()?;
Some((block_number, e.path()))
})
.collect();
indexed_paths.sort_by_key(|(num, _)| *num);
for (block_number, path) in indexed_paths {
let content = std::fs::read_to_string(&path)
.wrap_err_with(|| format!("Failed to read {:?}", path))?;
let file: GasRampPayloadFile = serde_json::from_str(&content)
.wrap_err_with(|| format!("Failed to parse {:?}", path))?;
let version = match file.version {
1 => EngineApiMessageVersion::V1,
2 => EngineApiMessageVersion::V2,
3 => EngineApiMessageVersion::V3,
4 => EngineApiMessageVersion::V4,
5 => EngineApiMessageVersion::V5,
v => return Err(eyre::eyre!("Invalid version {} in {:?}", v, path)),
};
info!(
block_number,
block_hash = %file.block_hash,
path = %path.display(),
"Loaded gas ramp payload"
);
payloads.push(GasRampPayload { block_number, version, file });
}
Ok(payloads)
}
async fn execute_payload_v4(
&self,
provider: &RootProvider<AnyNetwork>,
envelope: &ExecutionPayloadEnvelopeV4,
parent_hash: B256,
) -> eyre::Result<()> {
let block_hash =
envelope.envelope_inner.execution_payload.payload_inner.payload_inner.block_hash;
debug!(
method = "engine_newPayloadV4",
block_hash = %block_hash,
"Sending newPayload"
);
let status = provider
.new_payload_v4(
envelope.envelope_inner.execution_payload.clone(),
vec![],
B256::ZERO,
envelope.execution_requests.to_vec(),
)
.await?;
info!(?status, "newPayloadV4 response");
if !status.is_valid() {
return Err(eyre::eyre!("Payload rejected: {:?}", status));
}
let fcu_state = ForkchoiceState {
head_block_hash: block_hash,
safe_block_hash: parent_hash,
finalized_block_hash: parent_hash,
};
debug!(method = "engine_forkchoiceUpdatedV3", ?fcu_state, "Sending forkchoiceUpdated");
let fcu_result = provider.fork_choice_updated_v3(fcu_state, None).await?;
info!(?fcu_result, "forkchoiceUpdatedV3 response");
Ok(())
}
}

View File

@@ -3,16 +3,15 @@
//! before sending additional calls.
use alloy_eips::eip7685::Requests;
use alloy_primitives::B256;
use alloy_provider::{ext::EngineApi, network::AnyRpcBlock, Network, Provider};
use alloy_rpc_types_engine::{
ExecutionPayload, ExecutionPayloadInputV2, ExecutionPayloadSidecar, ForkchoiceState,
ForkchoiceUpdated, PayloadAttributes, PayloadStatus,
ExecutionPayload, ExecutionPayloadInputV2, ForkchoiceState, ForkchoiceUpdated,
PayloadAttributes, PayloadStatus,
};
use alloy_transport::TransportResult;
use op_alloy_rpc_types_engine::OpExecutionPayloadV4;
use reth_node_api::EngineApiMessageVersion;
use tracing::{debug, error};
use tracing::error;
/// An extension trait for providers that implement the engine API, to wait for a VALID response.
#[async_trait::async_trait]
@@ -53,13 +52,6 @@ where
fork_choice_state: ForkchoiceState,
payload_attributes: Option<PayloadAttributes>,
) -> TransportResult<ForkchoiceUpdated> {
debug!(
method = "engine_forkchoiceUpdatedV1",
?fork_choice_state,
?payload_attributes,
"Sending forkchoiceUpdated"
);
let mut status =
self.fork_choice_updated_v1(fork_choice_state, payload_attributes.clone()).await?;
@@ -90,13 +82,6 @@ where
fork_choice_state: ForkchoiceState,
payload_attributes: Option<PayloadAttributes>,
) -> TransportResult<ForkchoiceUpdated> {
debug!(
method = "engine_forkchoiceUpdatedV2",
?fork_choice_state,
?payload_attributes,
"Sending forkchoiceUpdated"
);
let mut status =
self.fork_choice_updated_v2(fork_choice_state, payload_attributes.clone()).await?;
@@ -127,13 +112,6 @@ where
fork_choice_state: ForkchoiceState,
payload_attributes: Option<PayloadAttributes>,
) -> TransportResult<ForkchoiceUpdated> {
debug!(
method = "engine_forkchoiceUpdatedV3",
?fork_choice_state,
?payload_attributes,
"Sending forkchoiceUpdated"
);
let mut status =
self.fork_choice_updated_v3(fork_choice_state, payload_attributes.clone()).await?;
@@ -170,47 +148,33 @@ pub(crate) fn block_to_new_payload(
// Convert to execution payload
let (payload, sidecar) = ExecutionPayload::from_block_slow(&block);
payload_to_new_payload(payload, sidecar, is_optimism, block.withdrawals_root, None)
}
pub(crate) fn payload_to_new_payload(
payload: ExecutionPayload,
sidecar: ExecutionPayloadSidecar,
is_optimism: bool,
withdrawals_root: Option<B256>,
target_version: Option<EngineApiMessageVersion>,
) -> eyre::Result<(EngineApiMessageVersion, serde_json::Value)> {
let (version, params) = match payload {
ExecutionPayload::V3(payload) => {
let cancun = sidecar.cancun().unwrap();
if let Some(prague) = sidecar.prague() {
// Use target version if provided (for Osaka), otherwise default to V4
let version = target_version.unwrap_or(EngineApiMessageVersion::V4);
if is_optimism {
let withdrawals_root = withdrawals_root.ok_or_else(|| {
eyre::eyre!("Missing withdrawals root for Optimism payload")
})?;
(
version,
EngineApiMessageVersion::V4,
serde_json::to_value((
OpExecutionPayloadV4 { payload_inner: payload, withdrawals_root },
OpExecutionPayloadV4 {
payload_inner: payload,
withdrawals_root: block.withdrawals_root.unwrap(),
},
cancun.versioned_hashes.clone(),
cancun.parent_beacon_block_root,
Requests::default(),
))?,
)
} else {
// Extract actual Requests from RequestsOrHash
let requests = prague.requests.requests_hash();
(
version,
EngineApiMessageVersion::V4,
serde_json::to_value((
payload,
cancun.versioned_hashes.clone(),
cancun.parent_beacon_block_root,
requests,
prague.requests.requests_hash(),
))?,
)
}
@@ -253,8 +217,6 @@ pub(crate) async fn call_new_payload<N: Network, P: Provider<N>>(
) -> TransportResult<()> {
let method = version.method_name();
debug!(method, "Sending newPayload");
let mut status: PayloadStatus = provider.client().request(method, &params).await?;
while !status.is_valid() {
@@ -275,15 +237,12 @@ pub(crate) async fn call_new_payload<N: Network, P: Provider<N>>(
/// Calls the correct `engine_forkchoiceUpdated` method depending on the given
/// `EngineApiMessageVersion`, using the provided forkchoice state and payload attributes for the
/// actual engine api message call.
///
/// Note: For Prague (V4), we still use forkchoiceUpdatedV3 as there is no V4.
pub(crate) async fn call_forkchoice_updated<N, P: EngineApiValidWaitExt<N>>(
provider: P,
message_version: EngineApiMessageVersion,
forkchoice_state: ForkchoiceState,
payload_attributes: Option<PayloadAttributes>,
) -> TransportResult<ForkchoiceUpdated> {
// FCU V3 is used for both Cancun and Prague (there is no FCU V4)
match message_version {
EngineApiMessageVersion::V3 | EngineApiMessageVersion::V4 | EngineApiMessageVersion::V5 => {
provider.fork_choice_updated_v3_wait(forkchoice_state, payload_attributes).await

View File

@@ -81,16 +81,12 @@ backon.workspace = true
tempfile.workspace = true
[features]
default = ["jemalloc", "otlp", "otlp-logs", "reth-revm/portable", "js-tracer", "keccak-cache-global", "asm-keccak"]
default = ["jemalloc", "otlp", "reth-revm/portable", "js-tracer", "keccak-cache-global", "asm-keccak"]
otlp = [
"reth-ethereum-cli/otlp",
"reth-node-core/otlp",
]
otlp-logs = [
"reth-ethereum-cli/otlp-logs",
"reth-node-core/otlp-logs",
]
js-tracer = [
"reth-node-builder/js-tracer",
"reth-node-ethereum/js-tracer",

View File

@@ -192,10 +192,10 @@ impl DeferredTrieData {
);
// Only trigger COW clone if there's actually data to add.
if !sorted_hashed_state.is_empty() {
Arc::make_mut(&mut overlay.state).extend_ref_and_sort(&sorted_hashed_state);
Arc::make_mut(&mut overlay.state).extend_ref(&sorted_hashed_state);
}
if !sorted_trie_updates.is_empty() {
Arc::make_mut(&mut overlay.nodes).extend_ref_and_sort(&sorted_trie_updates);
Arc::make_mut(&mut overlay.nodes).extend_ref(&sorted_trie_updates);
}
overlay
}
@@ -242,13 +242,13 @@ impl DeferredTrieData {
for ancestor in ancestors {
let ancestor_data = ancestor.wait_cloned();
state_mut.extend_ref_and_sort(ancestor_data.hashed_state.as_ref());
nodes_mut.extend_ref_and_sort(ancestor_data.trie_updates.as_ref());
state_mut.extend_ref(ancestor_data.hashed_state.as_ref());
nodes_mut.extend_ref(ancestor_data.trie_updates.as_ref());
}
// Extend with current block's sorted data last (takes precedence)
state_mut.extend_ref_and_sort(sorted_hashed_state);
nodes_mut.extend_ref_and_sort(sorted_trie_updates);
state_mut.extend_ref(sorted_hashed_state);
nodes_mut.extend_ref(sorted_trie_updates);
overlay
}
@@ -287,11 +287,6 @@ impl DeferredTrieData {
&inputs.ancestors,
);
*state = DeferredState::Ready(computed.clone());
// Release lock before inputs (and its ancestors) drop to avoid holding it
// while their potential last Arc refs drop (which could trigger recursive locking)
drop(state);
computed
}
}
@@ -521,7 +516,7 @@ mod tests {
let hashed_state = Arc::new(HashedPostStateSorted::new(accounts, B256Map::default()));
let trie_updates = Arc::default();
let mut overlay = TrieInputSorted::default();
Arc::make_mut(&mut overlay.state).extend_ref_and_sort(hashed_state.as_ref());
Arc::make_mut(&mut overlay.state).extend_ref(hashed_state.as_ref());
DeferredTrieData::ready(ComputedTrieData {
hashed_state,

View File

@@ -10,15 +10,15 @@ use alloy_primitives::{map::HashMap, BlockNumber, TxHash, B256};
use parking_lot::RwLock;
use reth_chainspec::ChainInfo;
use reth_ethereum_primitives::EthPrimitives;
use reth_execution_types::{BlockExecutionOutput, BlockExecutionResult, Chain, ExecutionOutcome};
use reth_execution_types::{Chain, ExecutionOutcome};
use reth_metrics::{metrics::Gauge, Metrics};
use reth_primitives_traits::{
BlockBody as _, IndexedTx, NodePrimitives, RecoveredBlock, SealedBlock, SealedHeader,
SignedTransaction,
};
use reth_storage_api::StateProviderBox;
use reth_trie::{updates::TrieUpdatesSorted, HashedPostStateSorted, LazyTrieData, TrieInputSorted};
use std::{collections::BTreeMap, sync::Arc, time::Instant};
use reth_trie::{updates::TrieUpdatesSorted, HashedPostStateSorted, TrieInputSorted};
use std::{collections::BTreeMap, ops::Deref, sync::Arc, time::Instant};
use tokio::sync::{broadcast, watch};
/// Size of the broadcast channel used to notify canonical state events.
@@ -648,7 +648,7 @@ impl<N: NodePrimitives> BlockState<N> {
}
/// Returns the `Receipts` of executed block that determines the state.
pub fn receipts(&self) -> &Vec<N::Receipt> {
pub fn receipts(&self) -> &Vec<Vec<N::Receipt>> {
&self.block.execution_outcome().receipts
}
@@ -659,7 +659,15 @@ impl<N: NodePrimitives> BlockState<N> {
///
/// This clones the vector of receipts. To avoid it, use [`Self::executed_block_receipts_ref`].
pub fn executed_block_receipts(&self) -> Vec<N::Receipt> {
self.receipts().clone()
let receipts = self.receipts();
debug_assert!(
receipts.len() <= 1,
"Expected at most one block's worth of receipts, found {}",
receipts.len()
);
receipts.first().cloned().unwrap_or_default()
}
/// Returns a slice of `Receipt` of executed block that determines the state.
@@ -667,7 +675,15 @@ impl<N: NodePrimitives> BlockState<N> {
/// has only one element corresponding to the executed block associated to
/// the state.
pub fn executed_block_receipts_ref(&self) -> &[N::Receipt] {
self.receipts()
let receipts = self.receipts();
debug_assert!(
receipts.len() <= 1,
"Expected at most one block's worth of receipts, found {}",
receipts.len()
);
receipts.first().map(|receipts| receipts.deref()).unwrap_or_default()
}
/// Returns an iterator over __parent__ `BlockStates`.
@@ -751,7 +767,7 @@ pub struct ExecutedBlock<N: NodePrimitives = EthPrimitives> {
/// Recovered Block
pub recovered_block: Arc<RecoveredBlock<N::Block>>,
/// Block's execution outcome.
pub execution_output: Arc<BlockExecutionOutput<N::Receipt>>,
pub execution_output: Arc<ExecutionOutcome<N::Receipt>>,
/// Deferred trie data produced by execution.
///
/// This allows deferring the computation of the trie data which can be expensive.
@@ -763,15 +779,7 @@ impl<N: NodePrimitives> Default for ExecutedBlock<N> {
fn default() -> Self {
Self {
recovered_block: Default::default(),
execution_output: Arc::new(BlockExecutionOutput {
result: BlockExecutionResult {
receipts: Default::default(),
requests: Default::default(),
gas_used: 0,
blob_gas_used: 0,
},
state: Default::default(),
}),
execution_output: Default::default(),
trie_data: DeferredTrieData::ready(ComputedTrieData::default()),
}
}
@@ -792,7 +800,7 @@ impl<N: NodePrimitives> ExecutedBlock<N> {
/// payload builders). This is the safe default path.
pub fn new(
recovered_block: Arc<RecoveredBlock<N::Block>>,
execution_output: Arc<BlockExecutionOutput<N::Receipt>>,
execution_output: Arc<ExecutionOutcome<N::Receipt>>,
trie_data: ComputedTrieData,
) -> Self {
Self { recovered_block, execution_output, trie_data: DeferredTrieData::ready(trie_data) }
@@ -814,7 +822,7 @@ impl<N: NodePrimitives> ExecutedBlock<N> {
/// Use [`Self::new()`] instead when trie data is already computed and available immediately.
pub const fn with_deferred_trie_data(
recovered_block: Arc<RecoveredBlock<N::Block>>,
execution_output: Arc<BlockExecutionOutput<N::Receipt>>,
execution_output: Arc<ExecutionOutcome<N::Receipt>>,
trie_data: DeferredTrieData,
) -> Self {
Self { recovered_block, execution_output, trie_data }
@@ -834,7 +842,7 @@ impl<N: NodePrimitives> ExecutedBlock<N> {
/// Returns a reference to the block's execution outcome
#[inline]
pub fn execution_outcome(&self) -> &BlockExecutionOutput<N::Receipt> {
pub fn execution_outcome(&self) -> &ExecutionOutcome<N::Receipt> {
&self.execution_output
}
@@ -934,39 +942,37 @@ impl<N: NodePrimitives<SignedTx: SignedTransaction>> NewCanonicalChain<N> {
pub fn to_chain_notification(&self) -> CanonStateNotification<N> {
match self {
Self::Commit { new } => {
CanonStateNotification::Commit { new: Arc::new(Self::blocks_to_chain(new)) }
}
Self::Reorg { new, old } => CanonStateNotification::Reorg {
new: Arc::new(Self::blocks_to_chain(new)),
old: Arc::new(Self::blocks_to_chain(old)),
},
}
}
/// Converts a slice of executed blocks into a [`Chain`].
fn blocks_to_chain(blocks: &[ExecutedBlock<N>]) -> Chain<N> {
match blocks {
[] => Chain::default(),
[first, rest @ ..] => {
let mut chain = Chain::from_block(
first.recovered_block().clone(),
ExecutionOutcome::from((
first.execution_outcome().clone(),
first.block_number(),
)),
LazyTrieData::ready(first.hashed_state(), first.trie_updates()),
);
for exec in rest {
let new = Arc::new(new.iter().fold(Chain::default(), |mut chain, exec| {
chain.append_block(
exec.recovered_block().clone(),
ExecutionOutcome::from((
exec.execution_outcome().clone(),
exec.block_number(),
)),
LazyTrieData::ready(exec.hashed_state(), exec.trie_updates()),
exec.execution_outcome().clone(),
exec.trie_updates(),
exec.hashed_state(),
);
}
chain
chain
}));
CanonStateNotification::Commit { new }
}
Self::Reorg { new, old } => {
let new = Arc::new(new.iter().fold(Chain::default(), |mut chain, exec| {
chain.append_block(
exec.recovered_block().clone(),
exec.execution_outcome().clone(),
exec.trie_updates(),
exec.hashed_state(),
);
chain
}));
let old = Arc::new(old.iter().fold(Chain::default(), |mut chain, exec| {
chain.append_block(
exec.recovered_block().clone(),
exec.execution_outcome().clone(),
exec.trie_updates(),
exec.hashed_state(),
);
chain
}));
CanonStateNotification::Reorg { new, old }
}
}
}
@@ -1260,7 +1266,7 @@ mod tests {
let state = BlockState::new(block);
assert_eq!(state.receipts(), receipts.first().unwrap());
assert_eq!(state.receipts(), &receipts);
}
#[test]
@@ -1537,31 +1543,33 @@ mod tests {
let block2a =
test_block_builder.get_executed_block_with_number(2, block1.recovered_block.hash());
let sample_execution_outcome = ExecutionOutcome {
receipts: vec![vec![], vec![]],
requests: vec![Requests::default(), Requests::default()],
..Default::default()
};
// Test commit notification
let chain_commit = NewCanonicalChain::Commit { new: vec![block0.clone(), block1.clone()] };
// Build expected trie data map
let mut expected_trie_data = BTreeMap::new();
expected_trie_data
.insert(0, LazyTrieData::ready(block0.hashed_state(), block0.trie_updates()));
expected_trie_data
.insert(1, LazyTrieData::ready(block1.hashed_state(), block1.trie_updates()));
// Build expected trie updates map
let mut expected_trie_updates = BTreeMap::new();
expected_trie_updates.insert(0, block0.trie_updates());
expected_trie_updates.insert(1, block1.trie_updates());
// Build expected execution outcome (first_block matches first block number)
let commit_execution_outcome = ExecutionOutcome {
receipts: vec![vec![], vec![]],
requests: vec![Requests::default(), Requests::default()],
first_block: 0,
..Default::default()
};
// Build expected hashed state map
let mut expected_hashed_state = BTreeMap::new();
expected_hashed_state.insert(0, block0.hashed_state());
expected_hashed_state.insert(1, block1.hashed_state());
assert_eq!(
chain_commit.to_chain_notification(),
CanonStateNotification::Commit {
new: Arc::new(Chain::new(
vec![block0.recovered_block().clone(), block1.recovered_block().clone()],
commit_execution_outcome,
expected_trie_data,
sample_execution_outcome.clone(),
expected_trie_updates,
expected_hashed_state
))
}
);
@@ -1572,39 +1580,40 @@ mod tests {
old: vec![block1.clone(), block2.clone()],
};
// Build expected trie data for old chain
let mut old_trie_data = BTreeMap::new();
old_trie_data.insert(1, LazyTrieData::ready(block1.hashed_state(), block1.trie_updates()));
old_trie_data.insert(2, LazyTrieData::ready(block2.hashed_state(), block2.trie_updates()));
// Build expected trie updates for old chain
let mut old_trie_updates = BTreeMap::new();
old_trie_updates.insert(1, block1.trie_updates());
old_trie_updates.insert(2, block2.trie_updates());
// Build expected trie data for new chain
let mut new_trie_data = BTreeMap::new();
new_trie_data
.insert(1, LazyTrieData::ready(block1a.hashed_state(), block1a.trie_updates()));
new_trie_data
.insert(2, LazyTrieData::ready(block2a.hashed_state(), block2a.trie_updates()));
// Build expected trie updates for new chain
let mut new_trie_updates = BTreeMap::new();
new_trie_updates.insert(1, block1a.trie_updates());
new_trie_updates.insert(2, block2a.trie_updates());
// Build expected execution outcome for reorg chains (first_block matches first block
// number)
let reorg_execution_outcome = ExecutionOutcome {
receipts: vec![vec![], vec![]],
requests: vec![Requests::default(), Requests::default()],
first_block: 1,
..Default::default()
};
// Build expected hashed state for old chain
let mut old_hashed_state = BTreeMap::new();
old_hashed_state.insert(1, block1.hashed_state());
old_hashed_state.insert(2, block2.hashed_state());
// Build expected hashed state for new chain
let mut new_hashed_state = BTreeMap::new();
new_hashed_state.insert(1, block1a.hashed_state());
new_hashed_state.insert(2, block2a.hashed_state());
assert_eq!(
chain_reorg.to_chain_notification(),
CanonStateNotification::Reorg {
old: Arc::new(Chain::new(
vec![block1.recovered_block().clone(), block2.recovered_block().clone()],
reorg_execution_outcome.clone(),
old_trie_data,
sample_execution_outcome.clone(),
old_trie_updates,
old_hashed_state
)),
new: Arc::new(Chain::new(
vec![block1a.recovered_block().clone(), block2a.recovered_block().clone()],
reorg_execution_outcome,
new_trie_data,
sample_execution_outcome,
new_trie_updates,
new_hashed_state
))
}
);

View File

@@ -1,228 +0,0 @@
//! Lazy overlay computation for trie input.
//!
//! This module provides [`LazyOverlay`], a type that computes the [`TrieInputSorted`]
//! lazily on first access. This allows execution to start before the trie overlay
//! is fully computed.
use crate::DeferredTrieData;
use alloy_primitives::B256;
use reth_trie::{updates::TrieUpdatesSorted, HashedPostStateSorted, TrieInputSorted};
use std::sync::{Arc, OnceLock};
use tracing::{debug, trace};
/// Inputs captured for lazy overlay computation.
#[derive(Clone)]
struct LazyOverlayInputs {
/// The persisted ancestor hash (anchor) this overlay should be built on.
anchor_hash: B256,
/// Deferred trie data handles for all in-memory blocks (newest to oldest).
blocks: Vec<DeferredTrieData>,
}
/// Lazily computed trie overlay.
///
/// Captures the inputs needed to compute a [`TrieInputSorted`] and defers the actual
/// computation until first access. This is conceptually similar to [`DeferredTrieData`]
/// but for overlay computation.
///
/// # Fast Path vs Slow Path
///
/// - **Fast path**: If the tip block's cached `anchored_trie_input` is ready and its `anchor_hash`
/// matches our expected anchor, we can reuse it directly (O(1)).
/// - **Slow path**: Otherwise, we merge all ancestor blocks' trie data into a new overlay.
#[derive(Clone)]
pub struct LazyOverlay {
/// Computed result, cached after first access.
inner: Arc<OnceLock<TrieInputSorted>>,
/// Inputs for lazy computation.
inputs: LazyOverlayInputs,
}
impl std::fmt::Debug for LazyOverlay {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("LazyOverlay")
.field("anchor_hash", &self.inputs.anchor_hash)
.field("num_blocks", &self.inputs.blocks.len())
.field("computed", &self.inner.get().is_some())
.finish()
}
}
impl LazyOverlay {
/// Create a new lazy overlay with the given anchor hash and block handles.
///
/// # Arguments
///
/// * `anchor_hash` - The persisted ancestor hash this overlay is built on top of
/// * `blocks` - Deferred trie data handles for in-memory blocks (newest to oldest)
pub fn new(anchor_hash: B256, blocks: Vec<DeferredTrieData>) -> Self {
Self { inner: Arc::new(OnceLock::new()), inputs: LazyOverlayInputs { anchor_hash, blocks } }
}
/// Returns the anchor hash this overlay is built on.
pub const fn anchor_hash(&self) -> B256 {
self.inputs.anchor_hash
}
/// Returns the number of in-memory blocks this overlay covers.
pub const fn num_blocks(&self) -> usize {
self.inputs.blocks.len()
}
/// Returns true if the overlay has already been computed.
pub fn is_computed(&self) -> bool {
self.inner.get().is_some()
}
/// Returns the computed trie input, computing it if necessary.
///
/// The first call triggers computation (which may block waiting for deferred data).
/// Subsequent calls return the cached result immediately.
pub fn get(&self) -> &TrieInputSorted {
self.inner.get_or_init(|| self.compute())
}
/// Returns the overlay as (nodes, state) tuple for use with `OverlayStateProviderFactory`.
pub fn as_overlay(&self) -> (Arc<TrieUpdatesSorted>, Arc<HashedPostStateSorted>) {
let input = self.get();
(Arc::clone(&input.nodes), Arc::clone(&input.state))
}
/// Compute the trie input overlay.
fn compute(&self) -> TrieInputSorted {
let anchor_hash = self.inputs.anchor_hash;
let blocks = &self.inputs.blocks;
if blocks.is_empty() {
debug!(target: "chain_state::lazy_overlay", "No in-memory blocks, returning empty overlay");
return TrieInputSorted::default();
}
// Fast path: Check if tip block's overlay is ready and anchor matches.
// The tip block (first in list) has the cumulative overlay from all ancestors.
if let Some(tip) = blocks.first() {
let data = tip.wait_cloned();
if let Some(anchored) = &data.anchored_trie_input {
if anchored.anchor_hash == anchor_hash {
trace!(target: "chain_state::lazy_overlay", %anchor_hash, "Reusing tip block's cached overlay (fast path)");
return (*anchored.trie_input).clone();
}
debug!(
target: "chain_state::lazy_overlay",
computed_anchor = %anchored.anchor_hash,
%anchor_hash,
"Anchor mismatch, falling back to merge"
);
}
}
// Slow path: Merge all blocks' trie data into a new overlay.
debug!(target: "chain_state::lazy_overlay", num_blocks = blocks.len(), "Merging blocks (slow path)");
Self::merge_blocks(blocks)
}
/// Merge all blocks' trie data into a single [`TrieInputSorted`].
///
/// Blocks are ordered newest to oldest. Uses hybrid merge algorithm that
/// switches between `extend_ref` (small batches) and k-way merge (large batches).
fn merge_blocks(blocks: &[DeferredTrieData]) -> TrieInputSorted {
const MERGE_BATCH_THRESHOLD: usize = 64;
if blocks.is_empty() {
return TrieInputSorted::default();
}
// Single block: use its data directly (no allocation)
if blocks.len() == 1 {
let data = blocks[0].wait_cloned();
return TrieInputSorted {
state: data.hashed_state,
nodes: data.trie_updates,
prefix_sets: Default::default(),
};
}
if blocks.len() < MERGE_BATCH_THRESHOLD {
// Small k: extend_ref loop with Arc::make_mut is faster.
// Uses copy-on-write - only clones inner data if Arc has multiple refs.
// Iterate oldest->newest so newer values override older ones.
let mut blocks_iter = blocks.iter().rev();
let first = blocks_iter.next().expect("blocks is non-empty");
let data = first.wait_cloned();
let mut state = data.hashed_state;
let mut nodes = data.trie_updates;
for block in blocks_iter {
let block_data = block.wait_cloned();
Arc::make_mut(&mut state).extend_ref_and_sort(block_data.hashed_state.as_ref());
Arc::make_mut(&mut nodes).extend_ref_and_sort(block_data.trie_updates.as_ref());
}
TrieInputSorted { state, nodes, prefix_sets: Default::default() }
} else {
// Large k: k-way merge is faster (O(n log k)).
// Collect is unavoidable here - we need all data materialized for k-way merge.
let trie_data: Vec<_> = blocks.iter().map(|b| b.wait_cloned()).collect();
let merged_state = HashedPostStateSorted::merge_batch(
trie_data.iter().map(|d| d.hashed_state.as_ref()),
);
let merged_nodes =
TrieUpdatesSorted::merge_batch(trie_data.iter().map(|d| d.trie_updates.as_ref()));
TrieInputSorted {
state: Arc::new(merged_state),
nodes: Arc::new(merged_nodes),
prefix_sets: Default::default(),
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use reth_trie::{updates::TrieUpdates, HashedPostState};
fn empty_deferred(anchor: B256) -> DeferredTrieData {
DeferredTrieData::pending(
Arc::new(HashedPostState::default()),
Arc::new(TrieUpdates::default()),
anchor,
Vec::new(),
)
}
#[test]
fn empty_blocks_returns_default() {
let overlay = LazyOverlay::new(B256::ZERO, vec![]);
let result = overlay.get();
assert!(result.state.is_empty());
assert!(result.nodes.is_empty());
}
#[test]
fn single_block_uses_data_directly() {
let anchor = B256::random();
let deferred = empty_deferred(anchor);
let overlay = LazyOverlay::new(anchor, vec![deferred]);
assert!(!overlay.is_computed());
let _ = overlay.get();
assert!(overlay.is_computed());
}
#[test]
fn cached_after_first_access() {
let overlay = LazyOverlay::new(B256::ZERO, vec![]);
// First access computes
let _ = overlay.get();
assert!(overlay.is_computed());
// Second access uses cache
let _ = overlay.get();
assert!(overlay.is_computed());
}
}

View File

@@ -14,9 +14,6 @@ pub use in_memory::*;
mod deferred_trie;
pub use deferred_trie::*;
mod lazy_overlay;
pub use lazy_overlay::*;
mod noop;
mod chain_info;

View File

@@ -280,6 +280,7 @@ mod tests {
vec![block1.clone(), block2.clone()],
ExecutionOutcome::default(),
BTreeMap::new(),
BTreeMap::new(),
));
// Create a commit notification
@@ -318,11 +319,13 @@ mod tests {
vec![block1.clone()],
ExecutionOutcome::default(),
BTreeMap::new(),
BTreeMap::new(),
));
let new_chain = Arc::new(Chain::new(
vec![block2.clone(), block3.clone()],
ExecutionOutcome::default(),
BTreeMap::new(),
BTreeMap::new(),
));
// Create a reorg notification
@@ -388,6 +391,7 @@ mod tests {
vec![block1.clone(), block2.clone()],
execution_outcome,
BTreeMap::new(),
BTreeMap::new(),
));
// Create a commit notification containing the new chain segment.
@@ -445,8 +449,12 @@ mod tests {
ExecutionOutcome { receipts: old_receipts, ..Default::default() };
// Create an old chain segment to be reverted, containing `old_block1`.
let old_chain: Arc<Chain> =
Arc::new(Chain::new(vec![old_block1.clone()], old_execution_outcome, BTreeMap::new()));
let old_chain: Arc<Chain> = Arc::new(Chain::new(
vec![old_block1.clone()],
old_execution_outcome,
BTreeMap::new(),
BTreeMap::new(),
));
// Define block2 for the new chain segment, which will be committed.
let mut body = BlockBody::<TransactionSigned>::default();
@@ -474,8 +482,12 @@ mod tests {
ExecutionOutcome { receipts: new_receipts, ..Default::default() };
// Create a new chain segment to be committed, containing `new_block1`.
let new_chain =
Arc::new(Chain::new(vec![new_block1.clone()], new_execution_outcome, BTreeMap::new()));
let new_chain = Arc::new(Chain::new(
vec![new_block1.clone()],
new_execution_outcome,
BTreeMap::new(),
BTreeMap::new(),
));
// Create a reorg notification with both reverted (old) and committed (new) chain segments.
let notification = CanonStateNotification::Reorg { old: old_chain, new: new_chain };

View File

@@ -3,7 +3,10 @@ use crate::{
CanonStateSubscriptions, ComputedTrieData,
};
use alloy_consensus::{Header, SignableTransaction, TxEip1559, TxReceipt, EMPTY_ROOT_HASH};
use alloy_eips::eip1559::{ETHEREUM_BLOCK_GAS_LIMIT_30M, INITIAL_BASE_FEE};
use alloy_eips::{
eip1559::{ETHEREUM_BLOCK_GAS_LIMIT_30M, INITIAL_BASE_FEE},
eip7685::Requests,
};
use alloy_primitives::{Address, BlockNumber, B256, U256};
use alloy_signer::SignerSync;
use alloy_signer_local::PrivateKeySigner;
@@ -13,7 +16,7 @@ use reth_chainspec::{ChainSpec, EthereumHardfork, MIN_TRANSACTION_GAS};
use reth_ethereum_primitives::{
Block, BlockBody, EthPrimitives, Receipt, Transaction, TransactionSigned,
};
use reth_execution_types::{BlockExecutionOutput, BlockExecutionResult, Chain, ExecutionOutcome};
use reth_execution_types::{Chain, ExecutionOutcome};
use reth_primitives_traits::{
proofs::{calculate_receipt_root, calculate_transaction_root, calculate_withdrawals_root},
Account, NodePrimitives, Recovered, RecoveredBlock, SealedBlock, SealedHeader,
@@ -198,7 +201,7 @@ impl<N: NodePrimitives> TestBlockBuilder<N> {
fn get_executed_block(
&mut self,
block_number: BlockNumber,
mut receipts: Vec<Vec<Receipt>>,
receipts: Vec<Vec<Receipt>>,
parent_hash: B256,
) -> ExecutedBlock {
let block = self.generate_random_block(block_number, parent_hash);
@@ -206,15 +209,12 @@ impl<N: NodePrimitives> TestBlockBuilder<N> {
let trie_data = ComputedTrieData::default();
ExecutedBlock::new(
Arc::new(RecoveredBlock::new_sealed(block, senders)),
Arc::new(BlockExecutionOutput {
result: BlockExecutionResult {
receipts: receipts.pop().unwrap_or_default(),
requests: Default::default(),
gas_used: 0,
blob_gas_used: 0,
},
state: BundleState::default(),
}),
Arc::new(ExecutionOutcome::new(
BundleState::default(),
receipts,
block_number,
vec![Requests::default()],
)),
trie_data,
)
}

View File

@@ -460,18 +460,6 @@ impl ChainSpec {
pub fn builder() -> ChainSpecBuilder {
ChainSpecBuilder::default()
}
/// Map a chain ID to a known chain spec, if available.
pub fn from_chain_id(chain_id: u64) -> Option<Arc<Self>> {
match NamedChain::try_from(chain_id).ok()? {
NamedChain::Mainnet => Some(MAINNET.clone()),
NamedChain::Sepolia => Some(SEPOLIA.clone()),
NamedChain::Holesky => Some(HOLESKY.clone()),
NamedChain::Hoodi => Some(HOODI.clone()),
NamedChain::Dev => Some(DEV.clone()),
_ => None,
}
}
}
impl<H: BlockHeader> ChainSpec<H> {

View File

@@ -100,7 +100,7 @@ impl<N: NodeTypes> TableViewer<()> for ListTableViewer<'_, N> {
tx.disable_long_read_transaction_safety();
let table_db = tx.inner.open_db(Some(self.args.table.name())).wrap_err("Could not open db.")?;
let stats = tx.inner.db_stat(table_db.dbi()).wrap_err(format!("Could not find table: {}", self.args.table.name()))?;
let stats = tx.inner.db_stat(&table_db).wrap_err(format!("Could not find table: {}", self.args.table.name()))?;
let total_entries = stats.entries();
let final_entry_idx = total_entries.saturating_sub(1);
if self.args.skip > final_entry_idx {

View File

@@ -54,18 +54,18 @@ pub enum SetCommand {
#[clap(action(ArgAction::Set))]
value: bool,
},
/// Store storage history in rocksdb instead of MDBX
/// Store storages history in RocksDB instead of MDBX
StoragesHistory {
#[clap(action(ArgAction::Set))]
value: bool,
},
/// Store transaction hash to number mapping in rocksdb instead of MDBX
TransactionHashNumbers {
/// Store account history in RocksDB instead of MDBX
AccountHistory {
#[clap(action(ArgAction::Set))]
value: bool,
},
/// Store account history in rocksdb instead of MDBX
AccountHistory {
/// Store transaction hash numbers in RocksDB instead of MDBX
TxHashNumbers {
#[clap(action(ArgAction::Set))]
value: bool,
},
@@ -151,14 +151,6 @@ impl Command {
settings.storages_history_in_rocksdb = value;
println!("Set storages_history_in_rocksdb = {}", value);
}
SetCommand::TransactionHashNumbers { value } => {
if settings.transaction_hash_numbers_in_rocksdb == value {
println!("transaction_hash_numbers_in_rocksdb is already set to {}", value);
return Ok(());
}
settings.transaction_hash_numbers_in_rocksdb = value;
println!("Set transaction_hash_numbers_in_rocksdb = {}", value);
}
SetCommand::AccountHistory { value } => {
if settings.account_history_in_rocksdb == value {
println!("account_history_in_rocksdb is already set to {}", value);
@@ -167,6 +159,14 @@ impl Command {
settings.account_history_in_rocksdb = value;
println!("Set account_history_in_rocksdb = {}", value);
}
SetCommand::TxHashNumbers { value } => {
if settings.transaction_hash_numbers_in_rocksdb == value {
println!("transaction_hash_numbers_in_rocksdb is already set to {}", value);
return Ok(());
}
settings.transaction_hash_numbers_in_rocksdb = value;
println!("Set transaction_hash_numbers_in_rocksdb = {}", value);
}
}
// Write updated settings

View File

@@ -88,7 +88,7 @@ impl Command {
let stats = tx
.inner
.db_stat(table_db.dbi())
.db_stat(&table_db)
.wrap_err(format!("Could not find table: {db_table}"))?;
// Defaults to 16KB right now but we should
@@ -129,8 +129,7 @@ impl Command {
table.add_row(row);
let freelist = tx.inner.env().freelist()?;
let pagesize =
tx.inner.db_stat(mdbx::Database::freelist_db().dbi())?.page_size() as usize;
let pagesize = tx.inner.db_stat(&mdbx::Database::freelist_db())?.page_size() as usize;
let freelist_size = freelist * pagesize;
let mut row = Row::new();

View File

@@ -2,15 +2,14 @@ use crate::common::EnvironmentArgs;
use clap::Parser;
use eyre::Result;
use lz4::Decoder;
use reqwest::{blocking::Client as BlockingClient, header::RANGE, Client, StatusCode};
use reqwest::Client;
use reth_chainspec::{EthChainSpec, EthereumHardforks};
use reth_cli::chainspec::ChainSpecParser;
use reth_fs_util as fs;
use std::{
borrow::Cow,
fs::OpenOptions,
io::{self, BufWriter, Read, Write},
path::{Path, PathBuf},
io::{self, Read, Write},
path::Path,
sync::{Arc, OnceLock},
time::{Duration, Instant},
};
@@ -87,9 +86,6 @@ impl DownloadDefaults {
"\nIf no URL is provided, the latest mainnet archive snapshot\nwill be proposed for download from ",
);
help.push_str(self.default_base_url.as_ref());
help.push_str(
".\n\nLocal file:// URLs are also supported for extracting snapshots from disk.",
);
help
}
@@ -297,14 +293,19 @@ impl CompressionFormat {
}
}
/// Extracts a compressed tar archive to the target directory with progress tracking.
fn extract_archive<R: Read>(
reader: R,
total_size: u64,
format: CompressionFormat,
target_dir: &Path,
) -> Result<()> {
let progress_reader = ProgressReader::new(reader, total_size);
/// Downloads and extracts a snapshot, blocking until finished.
fn blocking_download_and_extract(url: &str, target_dir: &Path) -> Result<()> {
let client = reqwest::blocking::Client::builder().build()?;
let response = client.get(url).send()?.error_for_status()?;
let total_size = response.content_length().ok_or_else(|| {
eyre::eyre!(
"Server did not provide Content-Length header. This is required for snapshot downloads"
)
})?;
let progress_reader = ProgressReader::new(response, total_size);
let format = CompressionFormat::from_url(url)?;
match format {
CompressionFormat::Lz4 => {
@@ -321,185 +322,6 @@ fn extract_archive<R: Read>(
Ok(())
}
/// Extracts a snapshot from a local file.
fn extract_from_file(path: &Path, format: CompressionFormat, target_dir: &Path) -> Result<()> {
let file = std::fs::File::open(path)?;
let total_size = file.metadata()?.len();
extract_archive(file, total_size, format, target_dir)
}
const MAX_DOWNLOAD_RETRIES: u32 = 10;
const RETRY_BACKOFF_SECS: u64 = 5;
/// Wrapper that tracks download progress while writing data.
/// Used with [`io::copy`] to display progress during downloads.
struct ProgressWriter<W> {
inner: W,
progress: DownloadProgress,
}
impl<W: Write> Write for ProgressWriter<W> {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
let n = self.inner.write(buf)?;
let _ = self.progress.update(n as u64);
Ok(n)
}
fn flush(&mut self) -> io::Result<()> {
self.inner.flush()
}
}
/// Downloads a file with resume support using HTTP Range requests.
/// Automatically retries on failure, resuming from where it left off.
/// Returns the path to the downloaded file and its total size.
fn resumable_download(url: &str, target_dir: &Path) -> Result<(PathBuf, u64)> {
let file_name = Url::parse(url)
.ok()
.and_then(|u| u.path_segments()?.next_back().map(|s| s.to_string()))
.unwrap_or_else(|| "snapshot.tar".to_string());
let final_path = target_dir.join(&file_name);
let part_path = target_dir.join(format!("{file_name}.part"));
let client = BlockingClient::builder().timeout(Duration::from_secs(30)).build()?;
let mut total_size: Option<u64> = None;
let mut last_error: Option<eyre::Error> = None;
for attempt in 1..=MAX_DOWNLOAD_RETRIES {
let existing_size = fs::metadata(&part_path).map(|m| m.len()).unwrap_or(0);
if let Some(total) = total_size &&
existing_size >= total
{
fs::rename(&part_path, &final_path)?;
info!(target: "reth::cli", "Download complete: {}", final_path.display());
return Ok((final_path, total));
}
if attempt > 1 {
info!(target: "reth::cli",
"Retry attempt {}/{} - resuming from {} bytes",
attempt, MAX_DOWNLOAD_RETRIES, existing_size
);
}
let mut request = client.get(url);
if existing_size > 0 {
request = request.header(RANGE, format!("bytes={existing_size}-"));
if attempt == 1 {
info!(target: "reth::cli", "Resuming download from {} bytes", existing_size);
}
}
let response = match request.send().and_then(|r| r.error_for_status()) {
Ok(r) => r,
Err(e) => {
last_error = Some(e.into());
if attempt < MAX_DOWNLOAD_RETRIES {
info!(target: "reth::cli",
"Download failed, retrying in {} seconds...", RETRY_BACKOFF_SECS
);
std::thread::sleep(Duration::from_secs(RETRY_BACKOFF_SECS));
}
continue;
}
};
let is_partial = response.status() == StatusCode::PARTIAL_CONTENT;
let size = if is_partial {
response
.headers()
.get("Content-Range")
.and_then(|v| v.to_str().ok())
.and_then(|v| v.split('/').next_back())
.and_then(|v| v.parse().ok())
} else {
response.content_length()
};
if total_size.is_none() {
total_size = size;
}
let current_total = total_size.ok_or_else(|| {
eyre::eyre!("Server did not provide Content-Length or Content-Range header")
})?;
let file = if is_partial && existing_size > 0 {
OpenOptions::new()
.append(true)
.open(&part_path)
.map_err(|e| fs::FsPathError::open(e, &part_path))?
} else {
fs::create_file(&part_path)?
};
let start_offset = if is_partial { existing_size } else { 0 };
let mut progress = DownloadProgress::new(current_total);
progress.downloaded = start_offset;
let mut writer = ProgressWriter { inner: BufWriter::new(file), progress };
let mut reader = response;
let copy_result = io::copy(&mut reader, &mut writer);
let flush_result = writer.inner.flush();
println!();
if let Err(e) = copy_result.and(flush_result) {
last_error = Some(e.into());
if attempt < MAX_DOWNLOAD_RETRIES {
info!(target: "reth::cli",
"Download interrupted, retrying in {} seconds...", RETRY_BACKOFF_SECS
);
std::thread::sleep(Duration::from_secs(RETRY_BACKOFF_SECS));
}
continue;
}
fs::rename(&part_path, &final_path)?;
info!(target: "reth::cli", "Download complete: {}", final_path.display());
return Ok((final_path, current_total));
}
Err(last_error
.unwrap_or_else(|| eyre::eyre!("Download failed after {} attempts", MAX_DOWNLOAD_RETRIES)))
}
/// Fetches the snapshot from a remote URL with resume support, then extracts it.
fn download_and_extract(url: &str, format: CompressionFormat, target_dir: &Path) -> Result<()> {
let (downloaded_path, total_size) = resumable_download(url, target_dir)?;
info!(target: "reth::cli", "Extracting snapshot...");
let file = fs::open(&downloaded_path)?;
extract_archive(file, total_size, format, target_dir)?;
fs::remove_file(&downloaded_path)?;
info!(target: "reth::cli", "Removed downloaded archive");
Ok(())
}
/// Downloads and extracts a snapshot, blocking until finished.
///
/// Supports both `file://` URLs for local files and HTTP(S) URLs for remote downloads.
fn blocking_download_and_extract(url: &str, target_dir: &Path) -> Result<()> {
let format = CompressionFormat::from_url(url)?;
if let Ok(parsed_url) = Url::parse(url) &&
parsed_url.scheme() == "file"
{
let file_path = parsed_url
.to_file_path()
.map_err(|_| eyre::eyre!("Invalid file:// URL path: {}", url))?;
extract_from_file(&file_path, format, target_dir)
} else {
download_and_extract(url, format, target_dir)
}
}
async fn stream_and_extract(url: &str, target_dir: &Path) -> Result<()> {
let target_dir = target_dir.to_path_buf();
let url = url.to_string();
@@ -558,7 +380,6 @@ mod tests {
assert!(help.contains("Available snapshot sources:"));
assert!(help.contains("merkle.io"));
assert!(help.contains("publicnode.com"));
assert!(help.contains("file://"));
}
#[test]
@@ -583,25 +404,4 @@ mod tests {
assert_eq!(defaults.available_snapshots.len(), 4); // 2 defaults + 2 added
assert_eq!(defaults.long_help, Some("Custom help for snapshots".to_string()));
}
#[test]
fn test_compression_format_detection() {
assert!(matches!(
CompressionFormat::from_url("https://example.com/snapshot.tar.lz4"),
Ok(CompressionFormat::Lz4)
));
assert!(matches!(
CompressionFormat::from_url("https://example.com/snapshot.tar.zst"),
Ok(CompressionFormat::Zstd)
));
assert!(matches!(
CompressionFormat::from_url("file:///path/to/snapshot.tar.lz4"),
Ok(CompressionFormat::Lz4)
));
assert!(matches!(
CompressionFormat::from_url("file:///path/to/snapshot.tar.zst"),
Ok(CompressionFormat::Zstd)
));
assert!(CompressionFormat::from_url("https://example.com/snapshot.tar.gz").is_err());
}
}

View File

@@ -42,9 +42,9 @@ pub struct Command<C: ChainSpecParser> {
#[arg(long)]
to: Option<u64>,
/// Number of tasks to run in parallel. Defaults to the number of available CPUs.
#[arg(long)]
num_tasks: Option<u64>,
/// Number of tasks to run in parallel
#[arg(long, default_value = "10")]
num_tasks: u64,
/// Continues with execution when an invalid block is encountered and collects these blocks.
#[arg(long)]
@@ -84,16 +84,12 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + Hardforks + EthereumHardforks>
}
};
let num_tasks = self.num_tasks.unwrap_or_else(|| {
std::thread::available_parallelism().map(|n| n.get() as u64).unwrap_or(10)
});
let total_blocks = max_block - min_block;
let total_gas = calculate_gas_used_from_headers(
&provider_factory.static_file_provider(),
min_block..=max_block,
)?;
let blocks_per_task = total_blocks / num_tasks;
let blocks_per_task = total_blocks / self.num_tasks;
let db_at = {
let provider_factory = provider_factory.clone();
@@ -111,10 +107,10 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + Hardforks + EthereumHardforks>
let _guard = cancellation.drop_guard();
let mut tasks = JoinSet::new();
for i in 0..num_tasks {
for i in 0..self.num_tasks {
let start_block = min_block + i * blocks_per_task;
let end_block =
if i == num_tasks - 1 { max_block } else { start_block + blocks_per_task };
if i == self.num_tasks - 1 { max_block } else { start_block + blocks_per_task };
// Spawn thread executing blocks
let provider_factory = provider_factory.clone();
@@ -152,7 +148,7 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + Hardforks + EthereumHardforks>
};
if let Err(err) = consensus
.validate_block_post_execution(&block, &result, None)
.validate_block_post_execution(&block, &result)
.wrap_err_with(|| {
format!("Failed to validate block {} {}", block.number(), block.hash())
})

View File

@@ -15,7 +15,7 @@ use reth_db_common::{
use reth_node_api::{HeaderTy, ReceiptTy, TxTy};
use reth_node_core::args::StageEnum;
use reth_provider::{
DBProvider, DatabaseProviderFactory, StaticFileProviderFactory, StaticFileWriter,
DBProvider, DatabaseProviderFactory, StaticFileProviderFactory, StaticFileWriter, TrieWriter,
};
use reth_prune::PruneSegment;
use reth_stages::StageId;
@@ -167,6 +167,10 @@ impl<C: ChainSpecParser> Command<C> {
None,
)?;
}
StageEnum::MerkleChangeSets => {
provider_rw.clear_trie_changesets()?;
reset_stage_checkpoint(tx, StageId::MerkleChangeSets)?;
}
StageEnum::AccountHistory | StageEnum::StorageHistory => {
tx.clear::<tables::AccountsHistory>()?;
tx.clear::<tables::StoragesHistory>()?;
@@ -178,6 +182,7 @@ impl<C: ChainSpecParser> Command<C> {
}
StageEnum::TxLookup => {
tx.clear::<tables::TransactionHashNumbers>()?;
reset_prune_checkpoint(tx, PruneSegment::TransactionLookup)?;
reset_stage_checkpoint(tx, StageId::TransactionLookup)?;

View File

@@ -42,7 +42,7 @@ where
Arc::new(output_db),
db_tool.chain(),
StaticFileProvider::read_write(output_datadir.static_files())?,
RocksDBProvider::builder(output_datadir.rocksdb()).build()?,
RocksDBProvider::builder(output_datadir.rocksdb()).with_default_tables().build()?,
)?,
to,
from,

View File

@@ -39,7 +39,7 @@ pub(crate) async fn dump_hashing_account_stage<N: ProviderNodeTypes<DB = Arc<Dat
Arc::new(output_db),
db_tool.chain(),
StaticFileProvider::read_write(output_datadir.static_files())?,
RocksDBProvider::builder(output_datadir.rocksdb()).build()?,
RocksDBProvider::builder(output_datadir.rocksdb()).with_default_tables().build()?,
)?,
to,
from,

View File

@@ -29,7 +29,7 @@ pub(crate) async fn dump_hashing_storage_stage<N: ProviderNodeTypes<DB = Arc<Dat
Arc::new(output_db),
db_tool.chain(),
StaticFileProvider::read_write(output_datadir.static_files())?,
RocksDBProvider::builder(output_datadir.rocksdb()).build()?,
RocksDBProvider::builder(output_datadir.rocksdb()).with_default_tables().build()?,
)?,
to,
from,

View File

@@ -62,7 +62,7 @@ where
Arc::new(output_db),
db_tool.chain(),
StaticFileProvider::read_write(output_datadir.static_files())?,
RocksDBProvider::builder(output_datadir.rocksdb()).build()?,
RocksDBProvider::builder(output_datadir.rocksdb()).with_default_tables().build()?,
)?,
to,
from,

View File

@@ -550,6 +550,7 @@ impl PruneConfig {
/// - `Option<PruneMode>` fields: set from `other` only if `self` is `None`.
/// - `block_interval`: set from `other` only if `self.block_interval ==
/// DEFAULT_BLOCK_INTERVAL`.
/// - `merkle_changesets`: always set from `other`.
/// - `receipts_log_filter`: set from `other` only if `self` is empty and `other` is non-empty.
pub fn merge(&mut self, other: Self) {
let Self {
@@ -562,6 +563,7 @@ impl PruneConfig {
account_history,
storage_history,
bodies_history,
merkle_changesets,
receipts_log_filter,
},
} = other;
@@ -578,6 +580,8 @@ impl PruneConfig {
self.segments.account_history = self.segments.account_history.or(account_history);
self.segments.storage_history = self.segments.storage_history.or(storage_history);
self.segments.bodies_history = self.segments.bodies_history.or(bodies_history);
// Merkle changesets is not optional; always take the value from `other`
self.segments.merkle_changesets = merkle_changesets;
if self.segments.receipts_log_filter.0.is_empty() && !receipts_log_filter.0.is_empty() {
self.segments.receipts_log_filter = receipts_log_filter;
@@ -1087,6 +1091,7 @@ receipts = { distance = 16384 }
account_history: None,
storage_history: Some(PruneMode::Before(5000)),
bodies_history: None,
merkle_changesets: PruneMode::Before(0),
receipts_log_filter: ReceiptsLogPruneConfig(BTreeMap::from([(
Address::random(),
PruneMode::Full,
@@ -1103,6 +1108,7 @@ receipts = { distance = 16384 }
account_history: Some(PruneMode::Distance(2000)),
storage_history: Some(PruneMode::Distance(3000)),
bodies_history: None,
merkle_changesets: PruneMode::Distance(10000),
receipts_log_filter: ReceiptsLogPruneConfig(BTreeMap::from([
(Address::random(), PruneMode::Distance(1000)),
(Address::random(), PruneMode::Before(2000)),
@@ -1121,6 +1127,7 @@ receipts = { distance = 16384 }
assert_eq!(config1.segments.receipts, Some(PruneMode::Distance(1000)));
assert_eq!(config1.segments.account_history, Some(PruneMode::Distance(2000)));
assert_eq!(config1.segments.storage_history, Some(PruneMode::Before(5000)));
assert_eq!(config1.segments.merkle_changesets, PruneMode::Distance(10000));
assert_eq!(config1.segments.receipts_log_filter, original_filter);
}

View File

@@ -15,12 +15,6 @@ use alloc::{boxed::Box, fmt::Debug, string::String, sync::Arc, vec::Vec};
use alloy_consensus::Header;
use alloy_primitives::{BlockHash, BlockNumber, Bloom, B256};
use core::error::Error;
/// Pre-computed receipt root and logs bloom.
///
/// When provided to [`FullConsensus::validate_block_post_execution`], this allows skipping
/// the receipt root computation and using the pre-computed values instead.
pub type ReceiptRootBloom = (B256, Bloom);
use reth_execution_types::BlockExecutionResult;
use reth_primitives_traits::{
constants::{GAS_LIMIT_BOUND_DIVISOR, MAXIMUM_GAS_LIMIT_BLOCK, MINIMUM_GAS_LIMIT},
@@ -45,15 +39,11 @@ pub trait FullConsensus<N: NodePrimitives>: Consensus<N::Block> {
///
/// See the Yellow Paper sections 4.3.2 "Holistic Validity".
///
/// If `receipt_root_bloom` is provided, the implementation should use the pre-computed
/// receipt root and logs bloom instead of computing them from the receipts.
///
/// Note: validating blocks does not include other validations of the Consensus
fn validate_block_post_execution(
&self,
block: &RecoveredBlock<N::Block>,
result: &BlockExecutionResult<N::Receipt>,
receipt_root_bloom: Option<ReceiptRootBloom>,
) -> Result<(), ConsensusError>;
}

View File

@@ -18,7 +18,7 @@
//!
//! **Not for production use** - provides no security guarantees or consensus validation.
use crate::{Consensus, ConsensusError, FullConsensus, HeaderValidator, ReceiptRootBloom};
use crate::{Consensus, ConsensusError, FullConsensus, HeaderValidator};
use alloc::sync::Arc;
use reth_execution_types::BlockExecutionResult;
use reth_primitives_traits::{Block, NodePrimitives, RecoveredBlock, SealedBlock, SealedHeader};
@@ -76,7 +76,6 @@ impl<N: NodePrimitives> FullConsensus<N> for NoopConsensus {
&self,
_block: &RecoveredBlock<N::Block>,
_result: &BlockExecutionResult<N::Receipt>,
_receipt_root_bloom: Option<ReceiptRootBloom>,
) -> Result<(), ConsensusError> {
Ok(())
}

View File

@@ -1,4 +1,4 @@
use crate::{Consensus, ConsensusError, FullConsensus, HeaderValidator, ReceiptRootBloom};
use crate::{Consensus, ConsensusError, FullConsensus, HeaderValidator};
use core::sync::atomic::{AtomicBool, Ordering};
use reth_execution_types::BlockExecutionResult;
use reth_primitives_traits::{Block, NodePrimitives, RecoveredBlock, SealedBlock, SealedHeader};
@@ -51,7 +51,6 @@ impl<N: NodePrimitives> FullConsensus<N> for TestConsensus {
&self,
_block: &RecoveredBlock<N::Block>,
_result: &BlockExecutionResult<N::Receipt>,
_receipt_root_bloom: Option<ReceiptRootBloom>,
) -> Result<(), ConsensusError> {
if self.fail_validation() {
Err(ConsensusError::BaseFeeMissing)

View File

@@ -125,7 +125,10 @@ pub async fn setup_engine_with_chain_import(
db.clone(),
chain_spec.clone(),
reth_provider::providers::StaticFileProvider::read_write(static_files_path.clone())?,
reth_provider::providers::RocksDBProvider::builder(rocksdb_dir_path).build().unwrap(),
reth_provider::providers::RocksDBProvider::builder(rocksdb_dir_path)
.with_default_tables()
.build()
.unwrap(),
)?;
// Initialize genesis if needed
@@ -328,6 +331,7 @@ mod tests {
reth_provider::providers::StaticFileProvider::read_write(static_files_path.clone())
.unwrap(),
reth_provider::providers::RocksDBProvider::builder(rocksdb_dir_path.clone())
.with_default_tables()
.build()
.unwrap(),
)
@@ -392,6 +396,7 @@ mod tests {
reth_provider::providers::StaticFileProvider::read_only(static_files_path, false)
.unwrap(),
reth_provider::providers::RocksDBProvider::builder(rocksdb_dir_path)
.with_default_tables()
.build()
.unwrap(),
)
@@ -490,7 +495,10 @@ mod tests {
db.clone(),
chain_spec.clone(),
reth_provider::providers::StaticFileProvider::read_write(static_files_path).unwrap(),
reth_provider::providers::RocksDBProvider::builder(rocksdb_dir_path).build().unwrap(),
reth_provider::providers::RocksDBProvider::builder(rocksdb_dir_path)
.with_default_tables()
.build()
.unwrap(),
)
.expect("failed to create provider factory");

View File

@@ -448,14 +448,12 @@ mod tests {
nonce: account.nonce,
code_hash: account.bytecode_hash.unwrap_or_default(),
code: None,
account_id: None,
}),
original_info: (i == 0).then(|| AccountInfo {
balance: account.balance.checked_div(U256::from(2)).unwrap_or(U256::ZERO),
nonce: 0,
code_hash: account.bytecode_hash.unwrap_or_default(),
code: None,
account_id: None,
}),
storage,
status: AccountStatus::default(),

View File

@@ -25,7 +25,6 @@ reth-tasks.workspace = true
reth-node-types.workspace = true
reth-chainspec.workspace = true
reth-engine-primitives.workspace = true
reth-trie-db.workspace = true
# async
futures.workspace = true
@@ -41,8 +40,6 @@ reth-evm-ethereum.workspace = true
reth-exex-types.workspace = true
reth-primitives-traits.workspace = true
reth-node-ethereum.workspace = true
reth-trie-db.workspace = true
alloy-eips.workspace = true
tokio = { workspace = true, features = ["sync"] }
tokio-stream.workspace = true

View File

@@ -26,7 +26,6 @@ use reth_provider::{
use reth_prune::PrunerWithFactory;
use reth_stages_api::{MetricEventsSender, Pipeline};
use reth_tasks::TaskSpawner;
use reth_trie_db::ChangesetCache;
use std::{
pin::Pin,
sync::Arc,
@@ -85,7 +84,6 @@ where
tree_config: TreeConfig,
sync_metrics_tx: MetricEventsSender,
evm_config: C,
changeset_cache: ChangesetCache,
) -> Self
where
V: EngineValidator<N::Payload>,
@@ -111,7 +109,6 @@ where
tree_config,
engine_kind,
evm_config,
changeset_cache,
);
let engine_handler = EngineApiRequestHandler::new(to_tree_tx, from_tree);
@@ -159,7 +156,6 @@ mod tests {
};
use reth_prune::Pruner;
use reth_tasks::TokioTaskExecutor;
use reth_trie_db::ChangesetCache;
use std::sync::Arc;
use tokio::sync::{mpsc::unbounded_channel, watch};
use tokio_stream::wrappers::UnboundedReceiverStream;
@@ -192,8 +188,6 @@ mod tests {
let pruner = Pruner::new_with_factory(provider_factory.clone(), vec![], 0, 0, None, rx);
let evm_config = EthEvmConfig::new(chain_spec.clone());
let changeset_cache = ChangesetCache::new();
let engine_validator = BasicEngineValidator::new(
blockchain_db.clone(),
consensus.clone(),
@@ -201,7 +195,6 @@ mod tests {
engine_payload_validator,
TreeConfig::default(),
Box::new(NoopInvalidBlockHook::default()),
changeset_cache.clone(),
);
let (sync_metrics_tx, _sync_metrics_rx) = unbounded_channel();
@@ -221,7 +214,6 @@ mod tests {
TreeConfig::default(),
sync_metrics_tx,
evm_config,
changeset_cache,
);
}
}

View File

@@ -34,8 +34,6 @@ reth-trie-parallel.workspace = true
reth-trie-sparse = { workspace = true, features = ["std", "metrics"] }
reth-trie-sparse-parallel = { workspace = true, features = ["std"] }
reth-trie.workspace = true
reth-trie-common.workspace = true
reth-trie-db.workspace = true
# alloy
alloy-evm.workspace = true
@@ -96,7 +94,7 @@ reth-tracing.workspace = true
reth-node-ethereum.workspace = true
reth-e2e-test-utils.workspace = true
# revm
# alloy
revm-state.workspace = true
assert_matches.workspace = true
@@ -135,8 +133,6 @@ test-utils = [
"reth-static-file",
"reth-tracing",
"reth-trie/test-utils",
"reth-trie-common/test-utils",
"reth-trie-db/test-utils",
"reth-trie-sparse/test-utils",
"reth-prune-types?/test-utils",
"reth-trie-parallel/test-utils",

View File

@@ -26,9 +26,7 @@ fn create_bench_state(num_accounts: usize) -> EvmState {
nonce: 10,
code_hash: B256::from_slice(&rng.random::<[u8; 32]>()),
code: Default::default(),
account_id: None,
},
original_info: Box::new(AccountInfo::default()),
storage,
status: AccountStatus::empty(),
transaction_id: 0,

View File

@@ -62,7 +62,6 @@ fn create_bench_state_updates(params: &BenchParams) -> Vec<EvmState> {
storage: HashMap::default(),
status: AccountStatus::SelfDestructed,
transaction_id: 0,
original_info: Box::new(AccountInfo::default()),
}
} else {
RevmAccount {
@@ -71,7 +70,6 @@ fn create_bench_state_updates(params: &BenchParams) -> Vec<EvmState> {
nonce: rng.random::<u64>(),
code_hash: KECCAK_EMPTY,
code: Some(Default::default()),
account_id: None,
},
storage: (0..rng.random_range(0..=params.storage_slots_per_account))
.map(|_| {
@@ -86,7 +84,6 @@ fn create_bench_state_updates(params: &BenchParams) -> Vec<EvmState> {
})
.collect(),
status: AccountStatus::Touched,
original_info: Box::new(AccountInfo::default()),
transaction_id: 0,
}
};
@@ -242,10 +239,7 @@ fn bench_state_root(c: &mut Criterion) {
std::convert::identity,
),
StateProviderBuilder::new(provider.clone(), genesis_hash, None),
OverlayStateProviderFactory::new(
provider,
reth_trie_db::ChangesetCache::new(),
),
OverlayStateProviderFactory::new(provider),
&TreeConfig::default(),
None,
);

View File

@@ -159,7 +159,6 @@ where
self.metrics.save_blocks_block_count.record(block_count as f64);
self.metrics.save_blocks_duration_seconds.record(start_time.elapsed());
Ok(last_block)
}
}

View File

@@ -60,22 +60,16 @@ impl EngineApiMetrics {
///
/// This method updates metrics for execution time, gas usage, and the number
/// of accounts, storage slots and bytecodes loaded and updated.
///
/// The optional `on_receipt` callback is invoked after each transaction with the receipt
/// index and a reference to all receipts collected so far. This allows callers to stream
/// receipts to a background task for incremental receipt root computation.
pub(crate) fn execute_metered<E, DB, F>(
pub(crate) fn execute_metered<E, DB>(
&self,
executor: E,
mut transactions: impl Iterator<Item = Result<impl ExecutableTx<E>, BlockExecutionError>>,
transaction_count: usize,
state_hook: Box<dyn OnStateHook>,
mut on_receipt: F,
) -> Result<(BlockExecutionOutput<E::Receipt>, Vec<Address>), BlockExecutionError>
where
DB: alloy_evm::Database,
E: BlockExecutor<Evm: Evm<DB: BorrowMut<State<DB>>>, Transaction: SignedTransaction>,
F: FnMut(&[E::Receipt]),
{
// clone here is cheap, all the metrics are Option<Arc<_>>. additionally
// they are globally registered so that the data recorded in the hook will
@@ -101,21 +95,14 @@ impl EngineApiMetrics {
let tx = tx?;
senders.push(*tx.signer());
let span = debug_span!(
target: "engine::tree",
"execute tx",
tx_hash = ?tx.tx().tx_hash(),
gas_used = tracing::field::Empty,
);
let span =
debug_span!(target: "engine::tree", "execute tx", tx_hash=?tx.tx().tx_hash());
let enter = span.entered();
trace!(target: "engine::tree", "Executing transaction");
let start = Instant::now();
let gas_used = executor.execute_transaction(tx)?;
self.executor.transaction_execution_histogram.record(start.elapsed());
// Invoke callback with the latest receipt
on_receipt(executor.receipts());
// record the tx gas used
enter.record("gas_used", gas_used);
}
@@ -270,10 +257,7 @@ impl ForkchoiceUpdatedMetrics {
pub(crate) struct NewPayloadStatusMetrics {
/// Finish time of the latest new payload call.
#[metric(skip)]
pub(crate) latest_finish_at: Option<Instant>,
/// Start time of the latest new payload call.
#[metric(skip)]
pub(crate) latest_start_at: Option<Instant>,
pub(crate) latest_at: Option<Instant>,
/// The total count of new payload messages received.
pub(crate) new_payload_messages: Counter,
/// The total count of new payload messages that we responded to with
@@ -301,10 +285,6 @@ pub(crate) struct NewPayloadStatusMetrics {
pub(crate) new_payload_latency: Histogram,
/// Latency for the last new payload call.
pub(crate) new_payload_last: Gauge,
/// Time from previous payload finish to current payload start (idle time).
pub(crate) time_between_new_payloads: Histogram,
/// Time from previous payload start to current payload start (total interval).
pub(crate) new_payload_interval: Histogram,
}
impl NewPayloadStatusMetrics {
@@ -318,14 +298,7 @@ impl NewPayloadStatusMetrics {
let finish = Instant::now();
let elapsed = finish - start;
if let Some(prev_finish) = self.latest_finish_at {
self.time_between_new_payloads.record(start - prev_finish);
}
if let Some(prev_start) = self.latest_start_at {
self.new_payload_interval.record(start - prev_start);
}
self.latest_finish_at = Some(finish);
self.latest_start_at = Some(start);
self.latest_at = Some(finish);
match result {
Ok(outcome) => match outcome.outcome.status {
PayloadStatusEnum::Valid => {
@@ -361,6 +334,10 @@ pub(crate) struct BlockValidationMetrics {
pub(crate) state_root_histogram: Histogram,
/// Histogram of deferred trie computation duration.
pub(crate) deferred_trie_compute_duration: Histogram,
/// Histogram of time spent waiting for deferred trie data to become available.
pub(crate) deferred_trie_wait_duration: Histogram,
/// Trie input computation duration
pub(crate) trie_input_duration: Histogram,
/// Payload conversion and validation latency
pub(crate) payload_validation_duration: Gauge,
/// Histogram of payload validation latency
@@ -431,13 +408,12 @@ mod tests {
/// A simple mock executor for testing that doesn't require complex EVM setup
struct MockExecutor {
state: EvmState,
receipts: Vec<Receipt>,
hook: Option<Box<dyn OnStateHook>>,
}
impl MockExecutor {
fn new(state: EvmState) -> Self {
Self { state, receipts: vec![], hook: None }
Self { state, hook: None }
}
}
@@ -519,16 +495,12 @@ mod tests {
self.hook = hook;
}
fn evm_mut(&mut self) -> &mut Self::Evm {
panic!("Mock executor evm_mut() not implemented")
}
fn evm(&self) -> &Self::Evm {
panic!("Mock executor evm() not implemented")
}
fn receipts(&self) -> &[Self::Receipt] {
&self.receipts
fn evm_mut(&mut self) -> &mut Self::Evm {
panic!("Mock executor evm_mut() not implemented")
}
}
@@ -563,12 +535,11 @@ mod tests {
let executor = MockExecutor::new(state);
// This will fail to create the EVM but should still call the hook
let _result = metrics.execute_metered::<_, EmptyDB, _>(
let _result = metrics.execute_metered::<_, EmptyDB>(
executor,
input.clone_transactions_recovered().map(Ok::<_, BlockExecutionError>),
input.transaction_count(),
state_hook,
|_| {},
);
// Check if hook was called (it might not be if finish() fails early)
@@ -609,9 +580,7 @@ mod tests {
nonce: 10,
code_hash: B256::random(),
code: Default::default(),
account_id: None,
},
original_info: Box::new(AccountInfo::default()),
storage,
status: AccountStatus::default(),
transaction_id: 0,
@@ -623,12 +592,11 @@ mod tests {
let executor = MockExecutor::new(state);
// Execute (will fail but should still update some metrics)
let _result = metrics.execute_metered::<_, EmptyDB, _>(
let _result = metrics.execute_metered::<_, EmptyDB>(
executor,
input.clone_transactions_recovered().map(Ok::<_, BlockExecutionError>),
input.transaction_count(),
state_hook,
|_| {},
);
let snapshot = snapshotter.snapshot().into_vec();

View File

@@ -30,13 +30,11 @@ use reth_payload_primitives::{
};
use reth_primitives_traits::{NodePrimitives, RecoveredBlock, SealedBlock, SealedHeader};
use reth_provider::{
BlockExecutionOutput, BlockExecutionResult, BlockNumReader, BlockReader, ChangeSetReader,
DatabaseProviderFactory, HashedPostStateProvider, ProviderError, StageCheckpointReader,
StateProviderBox, StateProviderFactory, StateReader, TransactionVariant,
BlockReader, DatabaseProviderFactory, HashedPostStateProvider, ProviderError, StateProviderBox,
StateProviderFactory, StateReader, TransactionVariant, TrieReader,
};
use reth_revm::database::StateProviderDatabase;
use reth_stages_api::ControlFlow;
use reth_trie_db::ChangesetCache;
use revm::state::EvmState;
use state::TreeState;
use std::{fmt::Debug, ops, sync::Arc, time::Instant};
@@ -273,8 +271,6 @@ where
engine_kind: EngineApiKind,
/// The EVM configuration.
evm_config: C,
/// Changeset cache for in-memory trie changesets
changeset_cache: ChangesetCache,
}
impl<N, P: Debug, T: PayloadTypes + Debug, V: Debug, C> std::fmt::Debug
@@ -299,7 +295,6 @@ where
.field("metrics", &self.metrics)
.field("engine_kind", &self.engine_kind)
.field("evm_config", &self.evm_config)
.field("changeset_cache", &self.changeset_cache)
.finish()
}
}
@@ -312,12 +307,11 @@ where
+ StateProviderFactory
+ StateReader<Receipt = N::Receipt>
+ HashedPostStateProvider
+ TrieReader
+ Clone
+ 'static,
<P as DatabaseProviderFactory>::Provider: BlockReader<Block = N::Block, Header = N::BlockHeader>
+ StageCheckpointReader
+ ChangeSetReader
+ BlockNumReader,
<P as DatabaseProviderFactory>::Provider:
BlockReader<Block = N::Block, Header = N::BlockHeader>,
C: ConfigureEvm<Primitives = N> + 'static,
T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>,
V: EngineValidator<T>,
@@ -337,7 +331,6 @@ where
config: TreeConfig,
engine_kind: EngineApiKind,
evm_config: C,
changeset_cache: ChangesetCache,
) -> Self {
let (incoming_tx, incoming) = crossbeam_channel::unbounded();
@@ -358,7 +351,6 @@ where
incoming_tx,
engine_kind,
evm_config,
changeset_cache,
}
}
@@ -378,7 +370,6 @@ where
config: TreeConfig,
kind: EngineApiKind,
evm_config: C,
changeset_cache: ChangesetCache,
) -> (Sender<FromEngine<EngineApiRequest<T, N>, N::Block>>, UnboundedReceiver<EngineApiEvent<N>>)
{
let best_block_number = provider.best_block_number().unwrap_or(0);
@@ -410,7 +401,6 @@ where
config,
kind,
evm_config,
changeset_cache,
);
let incoming = task.incoming_tx.clone();
std::thread::Builder::new().name("Engine Task".to_string()).spawn(|| task.run()).unwrap();
@@ -1375,21 +1365,6 @@ where
debug!(target: "engine::tree", ?last_persisted_block_hash, ?last_persisted_block_number, elapsed=?start_time.elapsed(), "Finished persisting, calling finish");
self.persistence_state.finish(last_persisted_block_hash, last_persisted_block_number);
// Evict trie changesets for blocks below the finalized block, but keep at least 64 blocks
if let Some(finalized) = self.canonical_in_memory_state.get_finalized_num_hash() {
let min_threshold = last_persisted_block_number.saturating_sub(64);
let eviction_threshold = finalized.number.min(min_threshold);
debug!(
target: "engine::tree",
last_persisted = last_persisted_block_number,
finalized_number = finalized.number,
eviction_threshold,
"Evicting changesets below threshold"
);
self.changeset_cache.evict(eviction_threshold);
}
self.on_new_persisted_block()?;
Ok(())
}
@@ -1478,7 +1453,7 @@ where
self.metrics.engine.forkchoice_updated.update_response_metrics(
start,
&mut self.metrics.engine.new_payload.latest_finish_at,
&mut self.metrics.engine.new_payload.latest_at,
has_attrs,
&output,
);
@@ -1676,18 +1651,6 @@ where
)));
return Ok(());
}
} else {
// We don't have the head block or any of its ancestors buffered. Request
// a download for the head block which will then trigger further sync.
debug!(
target: "engine::tree",
head_hash = %sync_target_state.head_block_hash,
"Backfill complete but head block not buffered, requesting download"
);
self.emit_event(EngineApiEvent::Download(DownloadRequest::single_block(
sync_target_state.head_block_hash,
)));
return Ok(());
}
// try to close the gap by executing buffered blocks that are child blocks of the new head
@@ -1855,7 +1818,6 @@ where
/// or the database. If the required historical data (such as trie change sets) has been
/// pruned for a given block, this operation will return an error. On archive nodes, it
/// can retrieve any block.
#[instrument(level = "debug", target = "engine::tree", skip(self))]
fn canonical_block_by_hash(&self, hash: B256) -> ProviderResult<Option<ExecutedBlock<N>>> {
trace!(target: "engine::tree", ?hash, "Fetching executed block by hash");
// check memory first
@@ -1868,23 +1830,12 @@ where
.sealed_block_with_senders(hash.into(), TransactionVariant::WithHash)?
.ok_or_else(|| ProviderError::HeaderNotFound(hash.into()))?
.split_sealed();
let mut execution_output = self
let execution_output = self
.provider
.get_state(block.header().number())?
.ok_or_else(|| ProviderError::StateForNumberNotFound(block.header().number()))?;
let hashed_state = self.provider.hashed_post_state(execution_output.state());
debug!(
target: "engine::tree",
number = ?block.number(),
"computing block trie updates",
);
let db_provider = self.provider.database_provider_ro()?;
let trie_updates = reth_trie_db::compute_block_trie_updates(
&self.changeset_cache,
&db_provider,
block.number(),
)?;
let trie_updates = self.provider.get_block_trie_updates(block.number())?;
let sorted_hashed_state = Arc::new(hashed_state.into_sorted());
let sorted_trie_updates = Arc::new(trie_updates);
@@ -1892,19 +1843,9 @@ where
let trie_data =
ComputedTrieData::without_trie_input(sorted_hashed_state, sorted_trie_updates);
let execution_output = Arc::new(BlockExecutionOutput {
state: execution_output.bundle,
result: BlockExecutionResult {
receipts: execution_output.receipts.pop().unwrap_or_default(),
requests: execution_output.requests.pop().unwrap_or_default(),
gas_used: block.gas_used(),
blob_gas_used: block.blob_gas_used().unwrap_or_default(),
},
});
Ok(Some(ExecutedBlock::new(
Arc::new(RecoveredBlock::new_sealed(block, senders)),
execution_output,
Arc::new(execution_output),
trie_data,
)))
}

View File

@@ -101,7 +101,7 @@ impl<'a> Iterator for BALSlotIter<'a> {
return None;
}
return Some((address, StorageKey::from(slot)));
return Some((address, slot));
}
// Move to next account
@@ -177,11 +177,13 @@ where
let mut storage_map = HashedStorage::new(false);
for slot_changes in &account_changes.storage_changes {
let hashed_slot = keccak256(slot_changes.slot.to_be_bytes::<32>());
let hashed_slot = keccak256(slot_changes.slot);
// Get the last change for this slot
if let Some(last_change) = slot_changes.changes.last() {
storage_map.storage.insert(hashed_slot, last_change.new_value);
storage_map
.storage
.insert(hashed_slot, U256::from_be_bytes(last_change.new_value.0));
}
}
@@ -235,8 +237,8 @@ mod tests {
let provider = StateProviderTest::default();
let address = Address::random();
let slot = U256::random();
let value = U256::random();
let slot = StorageKey::random();
let value = B256::random();
let slot_changes = SlotChanges { slot, changes: vec![StorageChange::new(0, value)] };
@@ -256,10 +258,10 @@ mod tests {
assert!(result.storages.contains_key(&hashed_address));
let storage = result.storages.get(&hashed_address).unwrap();
let hashed_slot = keccak256(slot.to_be_bytes::<32>());
let hashed_slot = keccak256(slot);
let stored_value = storage.storage.get(&hashed_slot).unwrap();
assert_eq!(*stored_value, value);
assert_eq!(*stored_value, U256::from_be_bytes(value.0));
}
#[test]
@@ -390,15 +392,15 @@ mod tests {
let provider = StateProviderTest::default();
let address = Address::random();
let slot = U256::random();
let slot = StorageKey::random();
// Multiple changes to the same slot - should take the last one
let slot_changes = SlotChanges {
slot,
changes: vec![
StorageChange::new(0, U256::from(100)),
StorageChange::new(1, U256::from(200)),
StorageChange::new(2, U256::from(300)),
StorageChange::new(0, B256::from(U256::from(100).to_be_bytes::<32>())),
StorageChange::new(1, B256::from(U256::from(200).to_be_bytes::<32>())),
StorageChange::new(2, B256::from(U256::from(300).to_be_bytes::<32>())),
],
};
@@ -416,7 +418,7 @@ mod tests {
let hashed_address = keccak256(address);
let storage = result.storages.get(&hashed_address).unwrap();
let hashed_slot = keccak256(slot.to_be_bytes::<32>());
let hashed_slot = keccak256(slot);
let stored_value = storage.storage.get(&hashed_slot).unwrap();
@@ -436,15 +438,15 @@ mod tests {
address: addr1,
storage_changes: vec![
SlotChanges {
slot: U256::from(100),
changes: vec![StorageChange::new(0, U256::ZERO)],
slot: StorageKey::from(U256::from(100)),
changes: vec![StorageChange::new(0, B256::ZERO)],
},
SlotChanges {
slot: U256::from(101),
changes: vec![StorageChange::new(0, U256::ZERO)],
slot: StorageKey::from(U256::from(101)),
changes: vec![StorageChange::new(0, B256::ZERO)],
},
],
storage_reads: vec![U256::from(102)],
storage_reads: vec![StorageKey::from(U256::from(102))],
balance_changes: vec![],
nonce_changes: vec![],
code_changes: vec![],
@@ -454,10 +456,10 @@ mod tests {
let account2 = AccountChanges {
address: addr2,
storage_changes: vec![SlotChanges {
slot: U256::from(200),
changes: vec![StorageChange::new(0, U256::ZERO)],
slot: StorageKey::from(U256::from(200)),
changes: vec![StorageChange::new(0, B256::ZERO)],
}],
storage_reads: vec![U256::from(201)],
storage_reads: vec![StorageKey::from(U256::from(201))],
balance_changes: vec![],
nonce_changes: vec![],
code_changes: vec![],
@@ -468,15 +470,15 @@ mod tests {
address: addr3,
storage_changes: vec![
SlotChanges {
slot: U256::from(300),
changes: vec![StorageChange::new(0, U256::ZERO)],
slot: StorageKey::from(U256::from(300)),
changes: vec![StorageChange::new(0, B256::ZERO)],
},
SlotChanges {
slot: U256::from(301),
changes: vec![StorageChange::new(0, U256::ZERO)],
slot: StorageKey::from(U256::from(301)),
changes: vec![StorageChange::new(0, B256::ZERO)],
},
],
storage_reads: vec![U256::from(302)],
storage_reads: vec![StorageKey::from(U256::from(302))],
balance_changes: vec![],
nonce_changes: vec![],
code_changes: vec![],

View File

@@ -28,10 +28,10 @@ use reth_evm::{
ConfigureEvm, EvmEnvFor, ExecutableTxIterator, ExecutableTxTuple, OnStateHook, SpecFor,
TxEnvFor,
};
use reth_execution_types::ExecutionOutcome;
use reth_primitives_traits::NodePrimitives;
use reth_provider::{
BlockExecutionOutput, BlockReader, DatabaseProviderROFactory, StateProvider,
StateProviderFactory, StateReader,
BlockReader, DatabaseProviderROFactory, StateProvider, StateProviderFactory, StateReader,
};
use reth_revm::{db::BundleState, state::EvmState};
use reth_trie::{hashed_cursor::HashedCursorFactory, trie_cursor::TrieCursorFactory};
@@ -61,7 +61,6 @@ mod configured_sparse_trie;
pub mod executor;
pub mod multiproof;
pub mod prewarm;
pub mod receipt_root_task;
pub mod sparse_trie;
use configured_sparse_trie::ConfiguredSparseTrie;
@@ -666,15 +665,13 @@ impl<Tx, Err, R: Send + Sync + 'static> PayloadHandle<Tx, Err, R> {
/// Terminates the entire caching task.
///
/// If the [`BlockExecutionOutput`] is provided it will update the shared cache using its
/// If the [`ExecutionOutcome`] is provided it will update the shared cache using its
/// bundle state. Using `Arc<ExecutionOutcome>` allows sharing with the main execution
/// path without cloning the expensive `BundleState`.
///
/// Returns a sender for the channel that should be notified on block validation success.
pub(super) fn terminate_caching(
&mut self,
execution_outcome: Option<Arc<BlockExecutionOutput<R>>>,
) -> Option<mpsc::Sender<()>> {
execution_outcome: Option<Arc<ExecutionOutcome<R>>>,
) {
self.prewarm_handle.terminate_caching(execution_outcome)
}
@@ -710,21 +707,15 @@ impl<R: Send + Sync + 'static> CacheTaskHandle<R> {
/// Terminates the entire pre-warming task.
///
/// If the [`BlockExecutionOutput`] is provided it will update the shared cache using its
/// If the [`ExecutionOutcome`] is provided it will update the shared cache using its
/// bundle state. Using `Arc<ExecutionOutcome>` avoids cloning the expensive `BundleState`.
#[must_use = "sender must be used and notified on block validation success"]
pub(super) fn terminate_caching(
&mut self,
execution_outcome: Option<Arc<BlockExecutionOutput<R>>>,
) -> Option<mpsc::Sender<()>> {
execution_outcome: Option<Arc<ExecutionOutcome<R>>>,
) {
if let Some(tx) = self.to_prewarm_task.take() {
let (valid_block_tx, valid_block_rx) = mpsc::channel();
let event = PrewarmTaskEvent::Terminate { execution_outcome, valid_block_rx };
let event = PrewarmTaskEvent::Terminate { execution_outcome };
let _ = tx.send(event);
Some(valid_block_tx)
} else {
None
}
}
}
@@ -733,10 +724,7 @@ impl<R> Drop for CacheTaskHandle<R> {
fn drop(&mut self) {
// Ensure we always terminate on drop - send None without needing Send + Sync bounds
if let Some(tx) = self.to_prewarm_task.take() {
let _ = tx.send(PrewarmTaskEvent::Terminate {
execution_outcome: None,
valid_block_rx: mpsc::channel().1,
});
let _ = tx.send(PrewarmTaskEvent::Terminate { execution_outcome: None });
}
}
}
@@ -898,7 +886,6 @@ mod tests {
use reth_revm::db::BundleState;
use reth_testing_utils::generators;
use reth_trie::{test_utils::state_root, HashedPostState};
use reth_trie_db::ChangesetCache;
use revm_primitives::{Address, HashMap, B256, KECCAK_EMPTY, U256};
use revm_state::{AccountInfo, AccountStatus, EvmState, EvmStorageSlot};
use std::sync::Arc;
@@ -1071,9 +1058,7 @@ mod tests {
nonce: rng.random::<u64>(),
code_hash: KECCAK_EMPTY,
code: Some(Default::default()),
account_id: None,
},
original_info: Box::new(AccountInfo::default()),
storage,
status: AccountStatus::Touched,
transaction_id: 0,
@@ -1156,7 +1141,7 @@ mod tests {
std::convert::identity,
),
StateProviderBuilder::new(provider_factory.clone(), genesis_hash, None),
OverlayStateProviderFactory::new(provider_factory, ChangesetCache::new()),
OverlayStateProviderFactory::new(provider_factory),
&TreeConfig::default(),
None, // No BAL for test
);

View File

@@ -20,7 +20,6 @@ use reth_trie_parallel::{
AccountMultiproofInput, ProofResultContext, ProofResultMessage, ProofWorkerHandle,
},
};
use revm_primitives::map::{hash_map, B256Map};
use std::{collections::BTreeMap, sync::Arc, time::Instant};
use tracing::{debug, error, instrument, trace};
@@ -141,27 +140,22 @@ impl ProofSequencer {
/// Adds a proof with the corresponding state update and returns all sequential proofs and state
/// updates if we have a continuous sequence
fn add_proof(&mut self, sequence: u64, update: SparseTrieUpdate) -> Vec<SparseTrieUpdate> {
// Optimization: fast path for in-order delivery to avoid BTreeMap overhead.
// If this is the expected sequence, return it immediately without buffering.
if sequence == self.next_to_deliver {
let mut consecutive_proofs = Vec::with_capacity(1);
consecutive_proofs.push(update);
self.next_to_deliver += 1;
// Check if we have subsequent proofs in the pending buffer
while let Some(pending) = self.pending_proofs.remove(&self.next_to_deliver) {
consecutive_proofs.push(pending);
self.next_to_deliver += 1;
}
return consecutive_proofs;
}
if sequence > self.next_to_deliver {
if sequence >= self.next_to_deliver {
self.pending_proofs.insert(sequence, update);
}
Vec::new()
let mut consecutive_proofs = Vec::with_capacity(self.pending_proofs.len());
let mut current_sequence = self.next_to_deliver;
// keep collecting proofs and state updates as long as we have consecutive sequence numbers
while let Some(pending) = self.pending_proofs.remove(&current_sequence) {
consecutive_proofs.push(pending);
current_sequence += 1;
}
self.next_to_deliver += consecutive_proofs.len() as u64;
consecutive_proofs
}
/// Returns true if we still have pending proofs
@@ -615,19 +609,7 @@ impl MultiProofTask {
self.multi_added_removed_keys.touch_accounts(targets.keys().copied());
// Clone+Arc MultiAddedRemovedKeys for sharing with the dispatched multiproof tasks
let multi_added_removed_keys = Arc::new(MultiAddedRemovedKeys {
account: self.multi_added_removed_keys.account.clone(),
storages: targets
.keys()
.filter_map(|account| {
self.multi_added_removed_keys
.storages
.get(account)
.cloned()
.map(|keys| (*account, keys))
})
.collect(),
});
let multi_added_removed_keys = Arc::new(self.multi_added_removed_keys.clone());
self.metrics.prefetch_proof_targets_accounts_histogram.record(targets.len() as f64);
self.metrics
@@ -723,33 +705,7 @@ impl MultiProofTask {
}
// Clone+Arc MultiAddedRemovedKeys for sharing with the dispatched multiproof tasks
let multi_added_removed_keys = Arc::new(MultiAddedRemovedKeys {
account: self.multi_added_removed_keys.account.clone(),
storages: {
let mut storages = B256Map::with_capacity_and_hasher(
not_fetched_state_update.storages.len(),
Default::default(),
);
for account in not_fetched_state_update
.storages
.keys()
.chain(not_fetched_state_update.accounts.keys())
{
if let hash_map::Entry::Vacant(entry) = storages.entry(*account) {
entry.insert(
self.multi_added_removed_keys
.storages
.get(account)
.cloned()
.unwrap_or_default(),
);
}
}
storages
},
});
let multi_added_removed_keys = Arc::new(self.multi_added_removed_keys.clone());
let chunking_len = not_fetched_state_update.chunking_length();
let mut spawned_proof_targets = MultiProofTargets::default();
@@ -1323,10 +1279,9 @@ mod tests {
use reth_provider::{
providers::OverlayStateProviderFactory, test_utils::create_test_provider_factory,
BlockNumReader, BlockReader, ChangeSetReader, DatabaseProviderFactory, LatestStateProvider,
PruneCheckpointReader, StageCheckpointReader, StateProviderBox,
PruneCheckpointReader, StageCheckpointReader, StateProviderBox, TrieReader,
};
use reth_trie::MultiProof;
use reth_trie_db::ChangesetCache;
use reth_trie_parallel::proof_task::{ProofTaskCtx, ProofWorkerHandle};
use revm_primitives::{B256, U256};
use std::sync::{Arc, OnceLock};
@@ -1347,6 +1302,7 @@ mod tests {
where
F: DatabaseProviderFactory<
Provider: BlockReader
+ TrieReader
+ StageCheckpointReader
+ PruneCheckpointReader
+ ChangeSetReader
@@ -1356,8 +1312,7 @@ mod tests {
+ 'static,
{
let rt_handle = get_test_runtime_handle();
let changeset_cache = ChangesetCache::new();
let overlay_factory = OverlayStateProviderFactory::new(factory, changeset_cache);
let overlay_factory = OverlayStateProviderFactory::new(factory);
let task_ctx = ProofTaskCtx::new(overlay_factory);
let proof_handle = ProofWorkerHandle::new(rt_handle, task_ctx, 1, 1, false);
let (to_sparse_trie, _receiver) = std::sync::mpsc::channel();
@@ -1369,7 +1324,7 @@ mod tests {
fn create_cached_provider<F>(factory: F) -> CachedStateProvider<StateProviderBox>
where
F: DatabaseProviderFactory<
Provider: BlockReader + StageCheckpointReader + PruneCheckpointReader,
Provider: BlockReader + TrieReader + StageCheckpointReader + PruneCheckpointReader,
> + Clone
+ Send
+ 'static,
@@ -1816,9 +1771,7 @@ mod tests {
nonce: 1,
code_hash: Default::default(),
code: Default::default(),
account_id: None,
},
original_info: Box::new(revm_state::AccountInfo::default()),
transaction_id: Default::default(),
storage: Default::default(),
status: revm_state::AccountStatus::Touched,
@@ -1835,9 +1788,7 @@ mod tests {
nonce: 2,
code_hash: Default::default(),
code: Default::default(),
account_id: None,
},
original_info: Box::new(revm_state::AccountInfo::default()),
transaction_id: Default::default(),
storage: Default::default(),
status: revm_state::AccountStatus::Touched,
@@ -1939,9 +1890,7 @@ mod tests {
nonce: 1,
code_hash: Default::default(),
code: Default::default(),
account_id: None,
},
original_info: Box::new(revm_state::AccountInfo::default()),
transaction_id: Default::default(),
storage: Default::default(),
status: revm_state::AccountStatus::Touched,

View File

@@ -30,12 +30,10 @@ use alloy_primitives::{keccak256, map::B256Set, B256};
use crossbeam_channel::Sender as CrossbeamSender;
use metrics::{Counter, Gauge, Histogram};
use reth_evm::{execute::ExecutableTxFor, ConfigureEvm, Evm, EvmFor, SpecFor};
use reth_execution_types::ExecutionOutcome;
use reth_metrics::Metrics;
use reth_primitives_traits::NodePrimitives;
use reth_provider::{
AccountReader, BlockExecutionOutput, BlockReader, StateProvider, StateProviderFactory,
StateReader,
};
use reth_provider::{AccountReader, BlockReader, StateProvider, StateProviderFactory, StateReader};
use reth_revm::{database::StateProviderDatabase, state::EvmState};
use reth_trie::MultiProofTargets;
use std::{
@@ -261,11 +259,7 @@ where
///
/// This method is called from `run()` only after all execution tasks are complete.
#[instrument(level = "debug", target = "engine::tree::payload_processor::prewarm", skip_all)]
fn save_cache(
self,
execution_outcome: Arc<BlockExecutionOutput<N::Receipt>>,
valid_block_rx: mpsc::Receiver<()>,
) {
fn save_cache(self, execution_outcome: Arc<ExecutionOutcome<N::Receipt>>) {
let start = Instant::now();
let Self { execution_cache, ctx: PrewarmContext { env, metrics, saved_cache, .. }, .. } =
@@ -283,7 +277,7 @@ where
// Insert state into cache while holding the lock
// Access the BundleState through the shared ExecutionOutcome
if new_cache.cache().insert_state(&execution_outcome.state).is_err() {
if new_cache.cache().insert_state(execution_outcome.state()).is_err() {
// Clear the cache on error to prevent having a polluted cache
*cached = None;
debug!(target: "engine::caching", "cleared execution cache on update error");
@@ -292,11 +286,9 @@ where
new_cache.update_metrics();
if valid_block_rx.recv().is_ok() {
// Replace the shared cache with the new one; the previous cache (if any) is
// dropped.
*cached = Some(new_cache);
}
// Replace the shared cache with the new one; the previous cache (if any) is
// dropped.
*cached = Some(new_cache);
});
let elapsed = start.elapsed();
@@ -427,10 +419,9 @@ where
// completed executing a set of transactions
self.send_multi_proof_targets(proof_targets);
}
PrewarmTaskEvent::Terminate { execution_outcome, valid_block_rx } => {
PrewarmTaskEvent::Terminate { execution_outcome } => {
trace!(target: "engine::tree::payload_processor::prewarm", "Received termination signal");
final_execution_outcome =
Some(execution_outcome.map(|outcome| (outcome, valid_block_rx)));
final_execution_outcome = Some(execution_outcome);
if finished_execution {
// all tasks are done, we can exit, which will save caches and exit
@@ -455,8 +446,8 @@ where
debug!(target: "engine::tree::payload_processor::prewarm", "Completed prewarm execution");
// save caches and finish using the shared ExecutionOutcome
if let Some(Some((execution_outcome, valid_block_rx))) = final_execution_outcome {
self.save_cache(execution_outcome, valid_block_rx);
if let Some(Some(execution_outcome)) = final_execution_outcome {
self.save_cache(execution_outcome);
}
}
}
@@ -576,14 +567,9 @@ where
.entered();
txs.recv()
} {
let enter = debug_span!(
target: "engine::tree::payload_processor::prewarm",
"prewarm tx",
index,
tx_hash = %tx.tx().tx_hash(),
is_success = tracing::field::Empty,
)
.entered();
let enter =
debug_span!(target: "engine::tree::payload_processor::prewarm", "prewarm tx", index, tx_hash=%tx.tx().tx_hash())
.entered();
// create the tx env
let start = Instant::now();
@@ -824,12 +810,7 @@ pub(super) enum PrewarmTaskEvent<R> {
Terminate {
/// The final execution outcome. Using `Arc` allows sharing with the main execution
/// path without cloning the expensive `BundleState`.
execution_outcome: Option<Arc<BlockExecutionOutput<R>>>,
/// Receiver for the block validation result.
///
/// Cache saving is racing the state root validation. We optimistically construct the
/// updated cache but only save it once we know the block is valid.
valid_block_rx: mpsc::Receiver<()>,
execution_outcome: Option<Arc<ExecutionOutcome<R>>>,
},
/// The outcome of a pre-warm task
Outcome {

View File

@@ -1,250 +0,0 @@
//! Receipt root computation in a background task.
//!
//! This module provides a streaming receipt root builder that computes the receipt trie root
//! in a background thread. Receipts are sent via a channel with their index, and for each
//! receipt received, the builder incrementally flushes leaves to the underlying
//! [`OrderedTrieRootEncodedBuilder`] when possible. When the channel closes, the task returns the
//! computed root.
use alloy_eips::Encodable2718;
use alloy_primitives::{Bloom, B256};
use crossbeam_channel::Receiver;
use reth_primitives_traits::Receipt;
use reth_trie_common::ordered_root::OrderedTrieRootEncodedBuilder;
use tokio::sync::oneshot;
/// Receipt with index, ready to be sent to the background task for encoding and trie building.
#[derive(Debug, Clone)]
pub struct IndexedReceipt<R> {
/// The transaction index within the block.
pub index: usize,
/// The receipt.
pub receipt: R,
}
impl<R> IndexedReceipt<R> {
/// Creates a new indexed receipt.
#[inline]
pub const fn new(index: usize, receipt: R) -> Self {
Self { index, receipt }
}
}
/// Handle for running the receipt root computation in a background task.
///
/// This struct holds the channels needed to receive receipts and send the result.
/// Use [`Self::run`] to execute the computation (typically in a spawned blocking task).
#[derive(Debug)]
pub struct ReceiptRootTaskHandle<R> {
/// Receiver for indexed receipts.
receipt_rx: Receiver<IndexedReceipt<R>>,
/// Sender for the computed result.
result_tx: oneshot::Sender<(B256, Bloom)>,
}
impl<R: Receipt> ReceiptRootTaskHandle<R> {
/// Creates a new handle from the receipt receiver and result sender channels.
pub const fn new(
receipt_rx: Receiver<IndexedReceipt<R>>,
result_tx: oneshot::Sender<(B256, Bloom)>,
) -> Self {
Self { receipt_rx, result_tx }
}
/// Runs the receipt root computation, consuming the handle.
///
/// This method receives indexed receipts from the channel, encodes them,
/// and builds the trie incrementally. When all receipts have been received
/// (channel closed), it sends the result through the oneshot channel.
///
/// This is designed to be called inside a blocking task (e.g., via
/// `executor.spawn_blocking(move || handle.run(receipts_len))`).
///
/// # Arguments
///
/// * `receipts_len` - The total number of receipts expected. This is needed to correctly order
/// the trie keys according to RLP encoding rules.
///
/// # Panics
///
/// Panics if the number of receipts received doesn't match `receipts_len`.
pub fn run(self, receipts_len: usize) {
let mut builder = OrderedTrieRootEncodedBuilder::new(receipts_len);
let mut aggregated_bloom = Bloom::ZERO;
let mut encode_buf = Vec::new();
for indexed_receipt in self.receipt_rx {
let receipt_with_bloom = indexed_receipt.receipt.with_bloom_ref();
encode_buf.clear();
receipt_with_bloom.encode_2718(&mut encode_buf);
aggregated_bloom |= *receipt_with_bloom.bloom_ref();
builder.push_unchecked(indexed_receipt.index, &encode_buf);
}
let root = builder.finalize().expect("receipt root builder incomplete");
let _ = self.result_tx.send((root, aggregated_bloom));
}
}
#[cfg(test)]
mod tests {
use super::*;
use alloy_consensus::{proofs::calculate_receipt_root, TxReceipt};
use alloy_primitives::{b256, hex, Address, Bytes, Log};
use crossbeam_channel::bounded;
use reth_ethereum_primitives::{Receipt, TxType};
#[tokio::test]
async fn test_receipt_root_task_empty() {
let (_tx, rx) = bounded::<IndexedReceipt<Receipt>>(1);
let (result_tx, result_rx) = oneshot::channel();
drop(_tx);
let handle = ReceiptRootTaskHandle::new(rx, result_tx);
tokio::task::spawn_blocking(move || handle.run(0)).await.unwrap();
let (root, bloom) = result_rx.await.unwrap();
// Empty trie root
assert_eq!(root, reth_trie_common::EMPTY_ROOT_HASH);
assert_eq!(bloom, Bloom::ZERO);
}
#[tokio::test]
async fn test_receipt_root_task_single_receipt() {
let receipts: Vec<Receipt> = vec![Receipt::default()];
let (tx, rx) = bounded(1);
let (result_tx, result_rx) = oneshot::channel();
let receipts_len = receipts.len();
let handle = ReceiptRootTaskHandle::new(rx, result_tx);
let join_handle = tokio::task::spawn_blocking(move || handle.run(receipts_len));
for (i, receipt) in receipts.clone().into_iter().enumerate() {
tx.send(IndexedReceipt::new(i, receipt)).unwrap();
}
drop(tx);
join_handle.await.unwrap();
let (root, _bloom) = result_rx.await.unwrap();
// Verify against the standard calculation
let receipts_with_bloom: Vec<_> = receipts.iter().map(|r| r.with_bloom_ref()).collect();
let expected_root = calculate_receipt_root(&receipts_with_bloom);
assert_eq!(root, expected_root);
}
#[tokio::test]
async fn test_receipt_root_task_multiple_receipts() {
let receipts: Vec<Receipt> = vec![Receipt::default(); 5];
let (tx, rx) = bounded(4);
let (result_tx, result_rx) = oneshot::channel();
let receipts_len = receipts.len();
let handle = ReceiptRootTaskHandle::new(rx, result_tx);
let join_handle = tokio::task::spawn_blocking(move || handle.run(receipts_len));
for (i, receipt) in receipts.into_iter().enumerate() {
tx.send(IndexedReceipt::new(i, receipt)).unwrap();
}
drop(tx);
join_handle.await.unwrap();
let (root, bloom) = result_rx.await.unwrap();
// Verify against expected values from existing test
assert_eq!(
root,
b256!("0x61353b4fb714dc1fccacbf7eafc4273e62f3d1eed716fe41b2a0cd2e12c63ebc")
);
assert_eq!(
bloom,
Bloom::from(hex!("00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"))
);
}
#[tokio::test]
async fn test_receipt_root_matches_standard_calculation() {
// Create some receipts with actual data
let receipts = vec![
Receipt {
tx_type: TxType::Legacy,
cumulative_gas_used: 21000,
success: true,
logs: vec![],
},
Receipt {
tx_type: TxType::Eip1559,
cumulative_gas_used: 42000,
success: true,
logs: vec![Log {
address: Address::ZERO,
data: alloy_primitives::LogData::new_unchecked(vec![B256::ZERO], Bytes::new()),
}],
},
Receipt {
tx_type: TxType::Eip2930,
cumulative_gas_used: 63000,
success: false,
logs: vec![],
},
];
// Calculate expected values first (before we move receipts)
let receipts_with_bloom: Vec<_> = receipts.iter().map(|r| r.with_bloom_ref()).collect();
let expected_root = calculate_receipt_root(&receipts_with_bloom);
let expected_bloom =
receipts_with_bloom.iter().fold(Bloom::ZERO, |bloom, r| bloom | r.bloom_ref());
// Calculate using the task
let (tx, rx) = bounded(4);
let (result_tx, result_rx) = oneshot::channel();
let receipts_len = receipts.len();
let handle = ReceiptRootTaskHandle::new(rx, result_tx);
let join_handle = tokio::task::spawn_blocking(move || handle.run(receipts_len));
for (i, receipt) in receipts.into_iter().enumerate() {
tx.send(IndexedReceipt::new(i, receipt)).unwrap();
}
drop(tx);
join_handle.await.unwrap();
let (task_root, task_bloom) = result_rx.await.unwrap();
assert_eq!(task_root, expected_root);
assert_eq!(task_bloom, expected_bloom);
}
#[tokio::test]
async fn test_receipt_root_task_out_of_order() {
let receipts: Vec<Receipt> = vec![Receipt::default(); 5];
// Calculate expected values first (before we move receipts)
let receipts_with_bloom: Vec<_> = receipts.iter().map(|r| r.with_bloom_ref()).collect();
let expected_root = calculate_receipt_root(&receipts_with_bloom);
let (tx, rx) = bounded(4);
let (result_tx, result_rx) = oneshot::channel();
let receipts_len = receipts.len();
let handle = ReceiptRootTaskHandle::new(rx, result_tx);
let join_handle = tokio::task::spawn_blocking(move || handle.run(receipts_len));
// Send in reverse order to test out-of-order handling
for (i, receipt) in receipts.into_iter().enumerate().rev() {
tx.send(IndexedReceipt::new(i, receipt)).unwrap();
}
drop(tx);
join_handle.await.unwrap();
let (root, _bloom) = result_rx.await.unwrap();
assert_eq!(root, expected_root);
}
}

View File

@@ -15,11 +15,9 @@ use alloy_eip7928::BlockAccessList;
use alloy_eips::{eip1898::BlockWithParent, NumHash};
use alloy_evm::Evm;
use alloy_primitives::B256;
use crate::tree::payload_processor::receipt_root_task::{IndexedReceipt, ReceiptRootTaskHandle};
use rayon::prelude::*;
use reth_chain_state::{CanonicalInMemoryState, DeferredTrieData, ExecutedBlock, LazyOverlay};
use reth_consensus::{ConsensusError, FullConsensus, ReceiptRootBloom};
use reth_chain_state::{CanonicalInMemoryState, DeferredTrieData, ExecutedBlock};
use reth_consensus::{ConsensusError, FullConsensus};
use reth_engine_primitives::{
ConfigureEngineEvm, ExecutableTxIterator, ExecutionPayload, InvalidBlockHook, PayloadValidator,
};
@@ -37,13 +35,12 @@ use reth_primitives_traits::{
};
use reth_provider::{
providers::OverlayStateProviderFactory, BlockExecutionOutput, BlockNumReader, BlockReader,
ChangeSetReader, DatabaseProviderFactory, DatabaseProviderROFactory, HashedPostStateProvider,
ProviderError, PruneCheckpointReader, StageCheckpointReader, StateProvider,
StateProviderFactory, StateReader,
ChangeSetReader, DatabaseProviderFactory, DatabaseProviderROFactory, ExecutionOutcome,
HashedPostStateProvider, ProviderError, PruneCheckpointReader, StageCheckpointReader,
StateProvider, StateProviderFactory, StateReader, TrieReader,
};
use reth_revm::db::State;
use reth_trie::{updates::TrieUpdates, HashedPostState, StateRoot};
use reth_trie_db::ChangesetCache;
use reth_trie::{updates::TrieUpdates, HashedPostState, StateRoot, TrieInputSorted};
use reth_trie_parallel::root::{ParallelStateRoot, ParallelStateRootError};
use revm_primitives::Address;
use std::{
@@ -132,8 +129,6 @@ where
metrics: EngineApiMetrics,
/// Validator for the payload.
validator: V,
/// Changeset cache for in-memory trie changesets
changeset_cache: ChangesetCache,
}
impl<N, P, Evm, V> BasicEngineValidator<P, Evm, V>
@@ -141,6 +136,7 @@ where
N: NodePrimitives,
P: DatabaseProviderFactory<
Provider: BlockReader
+ TrieReader
+ StageCheckpointReader
+ PruneCheckpointReader
+ ChangeSetReader
@@ -164,7 +160,6 @@ where
validator: V,
config: TreeConfig,
invalid_block_hook: Box<dyn InvalidBlockHook<N>>,
changeset_cache: ChangesetCache,
) -> Self {
let precompile_cache_map = PrecompileCacheMap::default();
let payload_processor = PayloadProcessor::new(
@@ -184,7 +179,6 @@ where
invalid_block_hook,
metrics: EngineApiMetrics::default(),
validator,
changeset_cache,
}
}
@@ -369,6 +363,7 @@ where
}
let parent_hash = input.parent_hash();
let block_num_hash = input.num_hash();
trace!(target: "engine::tree::payload_validator", "Fetching block state provider");
let _enter =
@@ -423,23 +418,13 @@ where
.map_err(Box::<dyn std::error::Error + Send + Sync>::from))
.map(Arc::new);
// Create lazy overlay from ancestors - this doesn't block, allowing execution to start
// before the trie data is ready. The overlay will be computed on first access.
let (lazy_overlay, anchor_hash) = Self::get_parent_lazy_overlay(parent_hash, ctx.state());
// Create overlay factory for payload processor (StateRootTask path needs it for
// multiproofs)
let overlay_factory =
OverlayStateProviderFactory::new(self.provider.clone(), self.changeset_cache.clone())
.with_block_hash(Some(anchor_hash))
.with_lazy_overlay(lazy_overlay);
// Spawn the appropriate processor based on strategy
let mut handle = ensure_ok!(self.spawn_payload_processor(
env.clone(),
txs,
provider_builder,
overlay_factory.clone(),
parent_hash,
ctx.state(),
strategy,
block_access_list,
));
@@ -455,44 +440,19 @@ where
state_provider = Box::new(InstrumentedStateProvider::new(state_provider, "engine"));
}
// Execute the block and handle any execution errors.
// The receipt root task is spawned before execution and receives receipts incrementally
// as transactions complete, allowing parallel computation during execution.
let (output, senders, receipt_root_rx) =
match self.execute_block(state_provider, env, &input, &mut handle) {
Ok(output) => output,
Err(err) => return self.handle_execution_error(input, err, &parent_block),
};
// Execute the block and handle any execution errors
let (output, senders) = match self.execute_block(state_provider, env, &input, &mut handle) {
Ok(output) => output,
Err(err) => return self.handle_execution_error(input, err, &parent_block),
};
// After executing the block we can stop prewarming transactions
handle.stop_prewarming_execution();
// Create ExecutionOutcome early so we can terminate caching before validation and state
// root computation. Using Arc allows sharing with both the caching task and the deferred
// trie task without cloning the expensive BundleState.
let output = Arc::new(output);
// Terminate caching task early since execution is complete and caching is no longer
// needed. This frees up resources while state root computation continues.
let valid_block_tx = handle.terminate_caching(Some(output.clone()));
let block = self.convert_to_block(input)?.with_senders(senders);
// Wait for the receipt root computation to complete.
let receipt_root_bloom = Some(
receipt_root_rx
.blocking_recv()
.expect("receipt root task dropped sender without result"),
);
let hashed_state = ensure_ok_post_block!(
self.validate_post_execution(
&block,
&parent_block,
&output,
&mut ctx,
receipt_root_bloom
),
self.validate_post_execution(&block, &parent_block, &output, &mut ctx),
block
);
@@ -525,7 +485,11 @@ where
}
StateRootStrategy::Parallel => {
debug!(target: "engine::tree::payload_validator", "Using parallel state root algorithm");
match self.compute_state_root_parallel(overlay_factory.clone(), &hashed_state) {
match self.compute_state_root_parallel(
block.parent_hash(),
&hashed_state,
ctx.state(),
) {
Ok(result) => {
let elapsed = root_time.elapsed();
info!(
@@ -561,7 +525,7 @@ where
}
let (root, updates) = ensure_ok_post_block!(
self.compute_state_root_serial(overlay_factory.clone(), &hashed_state),
self.compute_state_root_serial(block.parent_hash(), &hashed_state, ctx.state()),
block
);
(root, updates, root_time.elapsed())
@@ -591,18 +555,14 @@ where
.into())
}
if let Some(valid_block_tx) = valid_block_tx {
let _ = valid_block_tx.send(());
}
// Create ExecutionOutcome and wrap in Arc for sharing with both the caching task
// and the deferred trie task. This avoids cloning the expensive BundleState.
let execution_outcome = Arc::new(ExecutionOutcome::from((output, block_num_hash.number)));
Ok(self.spawn_deferred_trie_task(
block,
output,
&ctx,
hashed_state,
trie_output,
overlay_factory,
))
// Terminate prewarming task with the shared execution outcome
handle.terminate_caching(Some(Arc::clone(&execution_outcome)));
Ok(self.spawn_deferred_trie_task(block, execution_outcome, &ctx, hashed_state, trie_output))
}
/// Return sealed block header from database or in-memory state by hash.
@@ -640,21 +600,13 @@ where
/// Executes a block with the given state provider
#[instrument(level = "debug", target = "engine::tree::payload_validator", skip_all)]
#[expect(clippy::type_complexity)]
fn execute_block<S, Err, T>(
&mut self,
state_provider: S,
env: ExecutionEnv<Evm>,
input: &BlockOrPayload<T>,
handle: &mut PayloadHandle<impl ExecutableTxFor<Evm>, Err, N::Receipt>,
) -> Result<
(
BlockExecutionOutput<N::Receipt>,
Vec<Address>,
tokio::sync::oneshot::Receiver<(B256, alloy_primitives::Bloom)>,
),
InsertBlockErrorKind,
>
) -> Result<(BlockExecutionOutput<N::Receipt>, Vec<Address>), InsertBlockErrorKind>
where
S: StateProvider + Send,
Err: core::error::Error + Send + Sync + 'static,
@@ -693,14 +645,6 @@ where
});
}
// Spawn background task to compute receipt root and logs bloom incrementally.
// Unbounded channel is used since tx count bounds capacity anyway (max ~30k txs per block).
let receipts_len = input.transaction_count();
let (receipt_tx, receipt_rx) = crossbeam_channel::unbounded();
let (result_tx, result_rx) = tokio::sync::oneshot::channel();
let task_handle = ReceiptRootTaskHandle::new(receipt_rx, result_tx);
self.payload_processor.executor().spawn_blocking(move || task_handle.run(receipts_len));
let execution_start = Instant::now();
let state_hook = Box::new(handle.state_hook());
let (output, senders) = self.metrics.execute_metered(
@@ -708,30 +652,15 @@ where
handle.iter_transactions().map(|res| res.map_err(BlockExecutionError::other)),
input.transaction_count(),
state_hook,
|receipts| {
// Send the latest receipt to the background task for incremental root computation.
// The receipt is cloned here; encoding happens in the background thread.
if let Some(receipt) = receipts.last() {
// Infer tx_index from the number of receipts collected so far
let tx_index = receipts.len() - 1;
let _ = receipt_tx.send(IndexedReceipt::new(tx_index, receipt.clone()));
}
},
)?;
drop(receipt_tx);
let execution_finish = Instant::now();
let execution_time = execution_finish.duration_since(execution_start);
debug!(target: "engine::tree::payload_validator", elapsed = ?execution_time, "Executed block");
Ok((output, senders, result_rx))
Ok((output, senders))
}
/// Compute state root for the given hashed post state in parallel.
///
/// Uses an overlay factory which provides the state of the parent block, along with the
/// [`HashedPostState`] containing the changes of this block, to compute the state root and
/// trie updates for this block.
///
/// # Returns
///
/// Returns `Ok(_)` if computed successfully.
@@ -739,39 +668,58 @@ where
#[instrument(level = "debug", target = "engine::tree::payload_validator", skip_all)]
fn compute_state_root_parallel(
&self,
overlay_factory: OverlayStateProviderFactory<P>,
parent_hash: B256,
hashed_state: &HashedPostState,
state: &EngineApiTreeState<N>,
) -> Result<(B256, TrieUpdates), ParallelStateRootError> {
// The `hashed_state` argument will be taken into account as part of the overlay, but we
let (mut input, block_hash) = self.compute_trie_input(parent_hash, state)?;
// Extend state overlay with current block's sorted state.
input.prefix_sets.extend(hashed_state.construct_prefix_sets());
let sorted_hashed_state = hashed_state.clone_into_sorted();
Arc::make_mut(&mut input.state).extend_ref(&sorted_hashed_state);
let TrieInputSorted { nodes, state, prefix_sets: prefix_sets_mut } = input;
let factory = OverlayStateProviderFactory::new(self.provider.clone())
.with_block_hash(Some(block_hash))
.with_trie_overlay(Some(nodes))
.with_hashed_state_overlay(Some(state));
// The `hashed_state` argument is already taken into account as part of the overlay, but we
// need to use the prefix sets which were generated from it to indicate to the
// ParallelStateRoot which parts of the trie need to be recomputed.
let prefix_sets = hashed_state.construct_prefix_sets().freeze();
let overlay_factory =
overlay_factory.with_extended_hashed_state_overlay(hashed_state.clone_into_sorted());
ParallelStateRoot::new(overlay_factory, prefix_sets).incremental_root_with_updates()
let prefix_sets = prefix_sets_mut.freeze();
ParallelStateRoot::new(factory, prefix_sets).incremental_root_with_updates()
}
/// Compute state root for the given hashed post state in serial.
///
/// Uses an overlay factory which provides the state of the parent block, along with the
/// [`HashedPostState`] containing the changes of this block, to compute the state root and
/// trie updates for this block.
fn compute_state_root_serial(
&self,
overlay_factory: OverlayStateProviderFactory<P>,
parent_hash: B256,
hashed_state: &HashedPostState,
state: &EngineApiTreeState<N>,
) -> ProviderResult<(B256, TrieUpdates)> {
// The `hashed_state` argument will be taken into account as part of the overlay, but we
// need to use the prefix sets which were generated from it to indicate to the
// StateRoot which parts of the trie need to be recomputed.
let prefix_sets = hashed_state.construct_prefix_sets().freeze();
let overlay_factory =
overlay_factory.with_extended_hashed_state_overlay(hashed_state.clone_into_sorted());
let (mut input, block_hash) = self.compute_trie_input(parent_hash, state)?;
let provider = overlay_factory.database_provider_ro()?;
// Extend state overlay with current block's sorted state.
input.prefix_sets.extend(hashed_state.construct_prefix_sets());
let sorted_hashed_state = hashed_state.clone_into_sorted();
Arc::make_mut(&mut input.state).extend_ref(&sorted_hashed_state);
let TrieInputSorted { nodes, state, .. } = input;
let prefix_sets = hashed_state.construct_prefix_sets();
let factory = OverlayStateProviderFactory::new(self.provider.clone())
.with_block_hash(Some(block_hash))
.with_trie_overlay(Some(nodes))
.with_hashed_state_overlay(Some(state));
let provider = factory.database_provider_ro()?;
Ok(StateRoot::new(&provider, &provider)
.with_prefix_sets(prefix_sets)
.with_prefix_sets(prefix_sets.freeze())
.root_with_updates()?)
}
@@ -781,9 +729,6 @@ where
/// - parent header validation
/// - post-execution consensus validation
/// - state-root based post-execution validation
///
/// If `receipt_root_bloom` is provided, it will be used instead of computing the receipt root
/// and logs bloom from the receipts.
#[instrument(level = "debug", target = "engine::tree::payload_validator", skip_all)]
fn validate_post_execution<T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>>(
&self,
@@ -791,7 +736,6 @@ where
parent_block: &SealedHeader<N::BlockHeader>,
output: &BlockExecutionOutput<N::Receipt>,
ctx: &mut TreeCtx<'_, N>,
receipt_root_bloom: Option<ReceiptRootBloom>,
) -> Result<HashedPostState, InsertBlockErrorKind>
where
V: PayloadValidator<T, Block = N::Block>,
@@ -818,9 +762,7 @@ where
let _enter =
debug_span!(target: "engine::tree::payload_validator", "validate_block_post_execution")
.entered();
if let Err(err) =
self.consensus.validate_block_post_execution(block, output, receipt_root_bloom)
{
if let Err(err) = self.consensus.validate_block_post_execution(block, output) {
// call post-block hook
self.on_invalid_block(parent_block, block, output, None, ctx.state_mut());
return Err(err.into())
@@ -860,11 +802,6 @@ where
///
/// The method handles strategy fallbacks if the preferred approach fails, ensuring
/// block execution always completes with a valid state root.
///
/// # Arguments
///
/// * `overlay_factory` - Pre-computed overlay factory for multiproof generation
/// (`StateRootTask`)
#[allow(clippy::too_many_arguments)]
#[instrument(
level = "debug",
@@ -877,7 +814,8 @@ where
env: ExecutionEnv<Evm>,
txs: T,
provider_builder: StateProviderBuilder<N, P>,
overlay_factory: OverlayStateProviderFactory<P>,
parent_hash: B256,
state: &EngineApiTreeState<N>,
strategy: StateRootStrategy,
block_access_list: Option<Arc<BlockAccessList>>,
) -> Result<
@@ -890,14 +828,32 @@ where
> {
match strategy {
StateRootStrategy::StateRootTask => {
// Compute trie input
let trie_input_start = Instant::now();
let (trie_input, block_hash) = self.compute_trie_input(parent_hash, state)?;
// Create OverlayStateProviderFactory with sorted trie data for multiproofs
let TrieInputSorted { nodes, state, .. } = trie_input;
let multiproof_provider_factory =
OverlayStateProviderFactory::new(self.provider.clone())
.with_block_hash(Some(block_hash))
.with_trie_overlay(Some(nodes))
.with_hashed_state_overlay(Some(state));
// Record trie input duration including OverlayStateProviderFactory setup
self.metrics
.block_validation
.trie_input_duration
.record(trie_input_start.elapsed().as_secs_f64());
let spawn_start = Instant::now();
// Use the pre-computed overlay factory for multiproofs
let handle = self.payload_processor.spawn(
env,
txs,
provider_builder,
overlay_factory,
multiproof_provider_factory,
&self.config,
block_access_list,
);
@@ -991,36 +947,99 @@ where
self.invalid_block_hook.on_invalid_block(parent_header, block, output, trie_updates);
}
/// Creates a [`LazyOverlay`] for the parent block without blocking.
/// Computes [`TrieInputSorted`] for the provided parent hash by combining database state
/// with in-memory overlays.
///
/// Returns a lazy overlay that will compute the trie input on first access, and the anchor
/// block hash (the highest persisted ancestor). This allows execution to start immediately
/// while the trie input computation is deferred until the overlay is actually needed.
/// The goal of this function is to take in-memory blocks and generate a [`TrieInputSorted`]
/// that extends from the highest persisted ancestor up through the parent. This enables state
/// root computation and proof generation without requiring all blocks to be persisted
/// first.
///
/// If parent is on disk (no in-memory blocks), returns `None` for the lazy overlay.
fn get_parent_lazy_overlay(
/// It works as follows:
/// 1. Collect in-memory overlay blocks using [`crate::tree::TreeState::blocks_by_hash`]. This
/// returns the highest persisted ancestor hash (`block_hash`) and the list of in-memory
/// blocks building on top of it.
/// 2. Fast path: If the tip in-memory block's trie input is already anchored to `block_hash`
/// (its `anchor_hash` matches `block_hash`), reuse it directly.
/// 3. Slow path: Build a new [`TrieInputSorted`] by aggregating the overlay blocks (from oldest
/// to newest) on top of the database state at `block_hash`.
#[instrument(
level = "debug",
target = "engine::tree::payload_validator",
skip_all,
fields(parent_hash)
)]
fn compute_trie_input(
&self,
parent_hash: B256,
state: &EngineApiTreeState<N>,
) -> (Option<LazyOverlay>, B256) {
let (anchor_hash, blocks) =
) -> ProviderResult<(TrieInputSorted, B256)> {
let wait_start = Instant::now();
let (block_hash, blocks) =
state.tree_state.blocks_by_hash(parent_hash).unwrap_or_else(|| (parent_hash, vec![]));
if blocks.is_empty() {
debug!(target: "engine::tree::payload_validator", "Parent found on disk, no lazy overlay needed");
return (None, anchor_hash);
// Fast path: if the tip block's anchor matches the persisted ancestor hash, reuse its
// TrieInput. This means the TrieInputSorted already aggregates all in-memory overlays
// from that ancestor, so we can avoid re-aggregation.
if let Some(tip_block) = blocks.first() {
let data = tip_block.trie_data();
if let (Some(anchor_hash), Some(trie_input)) =
(data.anchor_hash(), data.trie_input().cloned()) &&
anchor_hash == block_hash
{
trace!(target: "engine::tree::payload_validator", %block_hash,"Reusing trie input with matching anchor hash");
self.metrics
.block_validation
.deferred_trie_wait_duration
.record(wait_start.elapsed().as_secs_f64());
return Ok(((*trie_input).clone(), block_hash));
}
}
debug!(
target: "engine::tree::payload_validator",
%anchor_hash,
num_blocks = blocks.len(),
"Creating lazy overlay for in-memory blocks"
);
if blocks.is_empty() {
debug!(target: "engine::tree::payload_validator", "Parent found on disk");
} else {
debug!(target: "engine::tree::payload_validator", historical = ?block_hash, blocks = blocks.len(), "Parent found in memory");
}
// Extract deferred trie data handles (non-blocking)
let handles: Vec<DeferredTrieData> = blocks.iter().map(|b| b.trie_data_handle()).collect();
// Extend with contents of parent in-memory blocks directly in sorted form.
let input = Self::merge_overlay_trie_input(&blocks);
(Some(LazyOverlay::new(anchor_hash, handles)), anchor_hash)
self.metrics
.block_validation
.deferred_trie_wait_duration
.record(wait_start.elapsed().as_secs_f64());
Ok((input, block_hash))
}
/// Aggregates multiple in-memory blocks into a single [`TrieInputSorted`] by combining their
/// state changes.
///
/// The input `blocks` vector is ordered newest -> oldest (see `TreeState::blocks_by_hash`).
/// We iterate it in reverse so we start with the oldest block's trie data and extend forward
/// toward the newest, ensuring newer state takes precedence.
fn merge_overlay_trie_input(blocks: &[ExecutedBlock<N>]) -> TrieInputSorted {
let mut input = TrieInputSorted::default();
let mut blocks_iter = blocks.iter().rev().peekable();
if let Some(first) = blocks_iter.next() {
let data = first.trie_data();
input.state = data.hashed_state;
input.nodes = data.trie_updates;
// Only clone and mutate if there are more in-memory blocks.
if blocks_iter.peek().is_some() {
let state_mut = Arc::make_mut(&mut input.state);
let nodes_mut = Arc::make_mut(&mut input.nodes);
for block in blocks_iter {
let data = block.trie_data();
state_mut.extend_ref(data.hashed_state.as_ref());
nodes_mut.extend_ref(data.trie_updates.as_ref());
}
}
}
input
}
/// Spawns a background task to compute and sort trie data for the executed block.
@@ -1042,11 +1061,10 @@ where
fn spawn_deferred_trie_task(
&self,
block: RecoveredBlock<N::Block>,
execution_outcome: Arc<BlockExecutionOutput<N::Receipt>>,
execution_outcome: Arc<ExecutionOutcome<N::Receipt>>,
ctx: &TreeCtx<'_, N>,
hashed_state: HashedPostState,
trie_output: TrieUpdates,
overlay_factory: OverlayStateProviderFactory<P>,
) -> ExecutedBlock<N> {
// Capture parent hash and ancestor overlays for deferred trie input construction.
let (anchor_hash, overlay_blocks) = ctx
@@ -1070,21 +1088,9 @@ where
let deferred_handle_task = deferred_trie_data.clone();
let block_validation_metrics = self.metrics.block_validation.clone();
// Capture block info and cache handle for changeset computation
let block_hash = block.hash();
let block_number = block.number();
let changeset_cache = self.changeset_cache.clone();
// Spawn background task to compute trie data. Calling `wait_cloned` will compute from
// the stored inputs and cache the result, so subsequent calls return immediately.
let compute_trie_input_task = move || {
let _span = debug_span!(
target: "engine::tree::payload_validator",
"compute_trie_input_task",
block_number
)
.entered();
let result = panic::catch_unwind(AssertUnwindSafe(|| {
let compute_start = Instant::now();
let computed = deferred_handle_task.wait_cloned();
@@ -1107,40 +1113,6 @@ where
.anchored_overlay_hashed_state_size
.record(anchored.trie_input.state.total_len() as f64);
}
// Compute and cache changesets using the computed trie_updates
let changeset_start = Instant::now();
// Get a provider from the overlay factory for trie cursor access
let changeset_result =
overlay_factory.database_provider_ro().and_then(|provider| {
reth_trie::changesets::compute_trie_changesets(
&provider,
&computed.trie_updates,
)
.map_err(ProviderError::Database)
});
match changeset_result {
Ok(changesets) => {
debug!(
target: "engine::tree::changeset",
?block_number,
elapsed = ?changeset_start.elapsed(),
"Computed and caching changesets"
);
changeset_cache.insert(block_hash, block_number, Arc::new(changesets));
}
Err(e) => {
warn!(
target: "engine::tree::changeset",
?block_number,
?e,
"Failed to compute changesets in deferred trie task"
);
}
}
}));
if result.is_err() {
@@ -1237,6 +1209,7 @@ impl<N, Types, P, Evm, V> EngineValidator<Types> for BasicEngineValidator<P, Evm
where
P: DatabaseProviderFactory<
Provider: BlockReader
+ TrieReader
+ StageCheckpointReader
+ PruneCheckpointReader
+ ChangeSetReader
@@ -1289,7 +1262,7 @@ where
fn on_inserted_executed_block(&self, block: ExecutedBlock<N>) {
self.payload_processor.on_inserted_executed_block(
block.recovered_block.block_with_parent(),
&block.execution_output.state,
block.execution_output.state(),
);
}
}

View File

@@ -7,7 +7,6 @@ use crate::{
PersistTarget, TreeConfig,
},
};
use reth_trie_db::ChangesetCache;
use alloy_eips::eip1898::BlockWithParent;
use alloy_primitives::{
@@ -27,7 +26,7 @@ use reth_ethereum_engine_primitives::EthEngineTypes;
use reth_ethereum_primitives::{Block, EthPrimitives};
use reth_evm_ethereum::MockEvmConfig;
use reth_primitives_traits::Block as _;
use reth_provider::test_utils::MockEthProvider;
use reth_provider::{test_utils::MockEthProvider, ExecutionOutcome};
use std::{
collections::BTreeMap,
str::FromStr,
@@ -193,7 +192,6 @@ impl TestHarness {
let payload_builder = PayloadBuilderHandle::new(to_payload_service);
let evm_config = MockEvmConfig::default();
let changeset_cache = ChangesetCache::new();
let engine_validator = BasicEngineValidator::new(
provider.clone(),
consensus.clone(),
@@ -201,7 +199,6 @@ impl TestHarness {
payload_validator,
TreeConfig::default(),
Box::new(NoopInvalidBlockHook::default()),
changeset_cache.clone(),
);
let tree = EngineApiTreeHandler::new(
@@ -218,7 +215,6 @@ impl TestHarness {
TreeConfig::default().with_legacy_state_root(false).with_has_enough_parallelism(true),
EngineApiKind::Ethereum,
evm_config,
changeset_cache,
);
let block_builder = TestBlockBuilder::default().with_chain_spec((*chain_spec).clone());
@@ -392,7 +388,6 @@ impl ValidatorTestHarness {
let provider = harness.provider.clone();
let payload_validator = MockEngineValidator;
let evm_config = MockEvmConfig::default();
let changeset_cache = ChangesetCache::new();
let validator = BasicEngineValidator::new(
provider,
@@ -401,7 +396,6 @@ impl ValidatorTestHarness {
payload_validator,
TreeConfig::default(),
Box::new(NoopInvalidBlockHook::default()),
changeset_cache,
);
Self { harness, validator, metrics: TestMetrics::default() }
@@ -838,7 +832,7 @@ fn test_tree_state_on_new_head_deep_fork() {
for block in &chain_a {
test_harness.tree.state.tree_state.insert_executed(ExecutedBlock::new(
Arc::new(block.clone()),
Arc::new(BlockExecutionOutput::default()),
Arc::new(ExecutionOutcome::default()),
empty_trie_data(),
));
}
@@ -847,7 +841,7 @@ fn test_tree_state_on_new_head_deep_fork() {
for block in &chain_b {
test_harness.tree.state.tree_state.insert_executed(ExecutedBlock::new(
Arc::new(block.clone()),
Arc::new(BlockExecutionOutput::default()),
Arc::new(ExecutionOutcome::default()),
empty_trie_data(),
));
}
@@ -1008,15 +1002,6 @@ async fn test_engine_tree_live_sync_transition_required_blocks_requested() {
_ => panic!("Unexpected event: {event:#?}"),
}
// After backfill completes with head not buffered, we also request head download
let event = test_harness.from_tree_rx.recv().await.unwrap();
match event {
EngineApiEvent::Download(DownloadRequest::BlockSet(hash_set)) => {
assert_eq!(hash_set, HashSet::from_iter([main_chain_last_hash]));
}
_ => panic!("Unexpected event: {event:#?}"),
}
let _ = test_harness
.tree
.on_engine_message(FromEngine::DownloadedBlocks(vec![main_chain

View File

@@ -38,7 +38,6 @@ tempfile.workspace = true
default = []
otlp = ["reth-tracing/otlp", "reth-node-core/otlp"]
otlp-logs = ["reth-tracing/otlp-logs", "reth-node-core/otlp-logs"]
dev = ["reth-cli-commands/arbitrary"]

View File

@@ -19,7 +19,7 @@ use reth_db::DatabaseEnv;
use reth_node_api::NodePrimitives;
use reth_node_builder::{NodeBuilder, WithLaunchContext};
use reth_node_core::{
args::{LogArgs, OtlpInitStatus, OtlpLogsStatus, TraceArgs},
args::{LogArgs, OtlpInitStatus, TraceArgs},
version::version_metadata,
};
use reth_node_metrics::recorder::install_prometheus_recorder;
@@ -223,19 +223,16 @@ impl<
/// If file logging is enabled, this function returns a guard that must be kept alive to ensure
/// that all logs are flushed to disk.
///
/// If an OTLP endpoint is specified, it will export traces and logs to the configured
/// collector.
/// If an OTLP endpoint is specified, it will export metrics to the configured collector.
pub fn init_tracing(
&mut self,
runner: &CliRunner,
mut layers: Layers,
) -> eyre::Result<Option<FileWorkerGuard>> {
let otlp_status = runner.block_on(self.traces.init_otlp_tracing(&mut layers))?;
let otlp_logs_status = runner.block_on(self.traces.init_otlp_logs(&mut layers))?;
let guard = self.logs.init_tracing_with_layers(layers)?;
info!(target: "reth::cli", "Initialized tracing, debug log directory: {}", self.logs.log_file_directory);
match otlp_status {
OtlpInitStatus::Started(endpoint) => {
info!(target: "reth::cli", "Started OTLP {:?} tracing export to {endpoint}", self.traces.protocol);
@@ -246,16 +243,6 @@ impl<
OtlpInitStatus::Disabled => {}
}
match otlp_logs_status {
OtlpLogsStatus::Started(endpoint) => {
info!(target: "reth::cli", "Started OTLP {:?} logs export to {endpoint}", self.traces.protocol);
}
OtlpLogsStatus::NoFeature => {
warn!(target: "reth::cli", "Provided OTLP logs arguments do not have effect, compile with the `otlp-logs` feature")
}
OtlpLogsStatus::Disabled => {}
}
Ok(guard)
}
}

View File

@@ -15,7 +15,7 @@ use alloc::{fmt::Debug, sync::Arc};
use alloy_consensus::{constants::MAXIMUM_EXTRA_DATA_SIZE, EMPTY_OMMER_ROOT_HASH};
use alloy_eips::eip7840::BlobParams;
use reth_chainspec::{EthChainSpec, EthereumHardforks};
use reth_consensus::{Consensus, ConsensusError, FullConsensus, HeaderValidator, ReceiptRootBloom};
use reth_consensus::{Consensus, ConsensusError, FullConsensus, HeaderValidator};
use reth_consensus_common::validation::{
validate_4844_header_standalone, validate_against_parent_4844,
validate_against_parent_eip1559_base_fee, validate_against_parent_gas_limit,
@@ -74,15 +74,8 @@ where
&self,
block: &RecoveredBlock<N::Block>,
result: &BlockExecutionResult<N::Receipt>,
receipt_root_bloom: Option<ReceiptRootBloom>,
) -> Result<(), ConsensusError> {
validate_block_post_execution(
block,
&self.chain_spec,
&result.receipts,
&result.requests,
receipt_root_bloom,
)
validate_block_post_execution(block, &self.chain_spec, &result.receipts, &result.requests)
}
}

View File

@@ -12,15 +12,11 @@ use reth_primitives_traits::{
///
/// - Compares the receipts root in the block header to the block body
/// - Compares the gas used in the block header to the actual gas usage after execution
///
/// If `receipt_root_bloom` is provided, the pre-computed receipt root and logs bloom are used
/// instead of computing them from the receipts.
pub fn validate_block_post_execution<B, R, ChainSpec>(
block: &RecoveredBlock<B>,
chain_spec: &ChainSpec,
receipts: &[R],
requests: &Requests,
receipt_root_bloom: Option<(B256, Bloom)>,
) -> Result<(), ConsensusError>
where
B: Block,
@@ -41,26 +37,19 @@ where
// operation as hashing that is required for state root got calculated in every
// transaction This was replaced with is_success flag.
// See more about EIP here: https://eips.ethereum.org/EIPS/eip-658
if chain_spec.is_byzantium_active_at_block(block.header().number()) {
let result = if let Some((receipts_root, logs_bloom)) = receipt_root_bloom {
compare_receipts_root_and_logs_bloom(
receipts_root,
logs_bloom,
block.header().receipts_root(),
block.header().logs_bloom(),
)
} else {
verify_receipts(block.header().receipts_root(), block.header().logs_bloom(), receipts)
};
if let Err(error) = result {
let receipts = receipts
.iter()
.map(|r| Bytes::from(r.with_bloom_ref().encoded_2718()))
.collect::<Vec<_>>();
tracing::debug!(%error, ?receipts, "receipts verification failed");
return Err(error)
}
if chain_spec.is_byzantium_active_at_block(block.header().number()) &&
let Err(error) = verify_receipts(
block.header().receipts_root(),
block.header().logs_bloom(),
receipts,
)
{
let receipts = receipts
.iter()
.map(|r| Bytes::from(r.with_bloom_ref().encoded_2718()))
.collect::<Vec<_>>();
tracing::debug!(%error, ?receipts, "receipts verification failed");
return Err(error)
}
// Validate that the header requests hash matches the calculated requests hash

View File

@@ -188,7 +188,6 @@ where
block: &'a SealedBlock<Block>,
) -> Result<EthBlockExecutionCtx<'a>, Self::Error> {
Ok(EthBlockExecutionCtx {
tx_count_hint: Some(block.transaction_count()),
parent_hash: block.header().parent_hash,
parent_beacon_block_root: block.header().parent_beacon_block_root,
ommers: &block.body().ommers,
@@ -203,7 +202,6 @@ where
attributes: Self::NextBlockEnvCtx,
) -> Result<EthBlockExecutionCtx<'_>, Self::Error> {
Ok(EthBlockExecutionCtx {
tx_count_hint: None,
parent_hash: parent.hash(),
parent_beacon_block_root: attributes.parent_beacon_block_root,
ommers: &[],
@@ -240,9 +238,8 @@ where
revm_spec_by_timestamp_and_block_number(self.chain_spec(), timestamp, block_number);
// configure evm env based on parent block
let mut cfg_env = CfgEnv::new()
.with_chain_id(self.chain_spec().chain().id())
.with_spec_and_mainnet_gas_params(spec);
let mut cfg_env =
CfgEnv::new().with_chain_id(self.chain_spec().chain().id()).with_spec(spec);
if let Some(blob_params) = &blob_params {
cfg_env.set_max_blobs_per_tx(blob_params.max_blobs_per_tx);
@@ -283,7 +280,6 @@ where
payload: &'a ExecutionData,
) -> Result<ExecutionCtxFor<'a, Self>, Self::Error> {
Ok(EthBlockExecutionCtx {
tx_count_hint: Some(payload.payload.transactions().len()),
parent_hash: payload.parent_hash(),
parent_beacon_block_root: payload.sidecar.parent_beacon_block_root(),
ommers: &[],
@@ -411,7 +407,7 @@ mod tests {
let db = CacheDB::<EmptyDBTyped<ProviderError>>::default();
let evm_env = EvmEnv {
cfg_env: CfgEnv::new().with_spec_and_mainnet_gas_params(SpecId::CONSTANTINOPLE),
cfg_env: CfgEnv::new().with_spec(SpecId::CONSTANTINOPLE),
..Default::default()
};
@@ -478,7 +474,7 @@ mod tests {
let db = CacheDB::<EmptyDBTyped<ProviderError>>::default();
let evm_env = EvmEnv {
cfg_env: CfgEnv::new().with_spec_and_mainnet_gas_params(SpecId::CONSTANTINOPLE),
cfg_env: CfgEnv::new().with_spec(SpecId::CONSTANTINOPLE),
..Default::default()
};

View File

@@ -65,12 +65,7 @@ impl BlockExecutorFactory for MockEvmConfig {
DB: Database + 'a,
I: Inspector<<Self::EvmFactory as EvmFactory>::Context<&'a mut State<DB>>> + 'a,
{
MockExecutor {
result: self.exec_results.lock().pop().unwrap(),
evm,
hook: None,
receipts: Vec::new(),
}
MockExecutor { result: self.exec_results.lock().pop().unwrap(), evm, hook: None }
}
}
@@ -81,7 +76,6 @@ pub struct MockExecutor<'a, DB: Database, I> {
evm: EthEvm<&'a mut State<DB>, I, PrecompilesMap>,
#[debug(skip)]
hook: Option<Box<dyn reth_evm::OnStateHook>>,
receipts: Vec<Receipt>,
}
impl<'a, DB: Database, I: Inspector<EthEvmContext<&'a mut State<DB>>>> BlockExecutor
@@ -95,10 +89,6 @@ impl<'a, DB: Database, I: Inspector<EthEvmContext<&'a mut State<DB>>>> BlockExec
Ok(())
}
fn receipts(&self) -> &[Self::Receipt] {
&self.receipts
}
fn execute_transaction_without_commit(
&mut self,
_tx: impl ExecutableTx<Self>,

View File

@@ -38,7 +38,6 @@ fn create_database_with_beacon_root_contract() -> CacheDB<EmptyDB> {
code_hash: keccak256(BEACON_ROOTS_CODE.clone()),
nonce: 1,
code: Some(Bytecode::new_raw(BEACON_ROOTS_CODE.clone())),
account_id: None,
};
db.insert_account_info(BEACON_ROOTS_ADDRESS, beacon_root_contract_account);
@@ -54,7 +53,6 @@ fn create_database_with_withdrawal_requests_contract() -> CacheDB<EmptyDB> {
balance: U256::ZERO,
code_hash: keccak256(WITHDRAWAL_REQUEST_PREDEPLOY_CODE.clone()),
code: Some(Bytecode::new_raw(WITHDRAWAL_REQUEST_PREDEPLOY_CODE.clone())),
account_id: None,
};
db.insert_account_info(
@@ -341,7 +339,6 @@ fn create_database_with_block_hashes(latest_block: u64) -> CacheDB<EmptyDB> {
code_hash: keccak256(HISTORY_STORAGE_CODE.clone()),
code: Some(Bytecode::new_raw(HISTORY_STORAGE_CODE.clone())),
nonce: 1,
account_id: None,
};
db.insert_account_info(HISTORY_STORAGE_ADDRESS, blockhashes_contract_account);

View File

@@ -303,8 +303,6 @@ where
let eth_config =
EthConfigHandler::new(ctx.node.provider().clone(), ctx.node.evm_config().clone());
let testing_skip_invalid_transactions = ctx.config.rpc.testing_skip_invalid_transactions;
self.inner
.launch_add_ons_with(ctx, move |container| {
container.modules.merge_if_module_configured(
@@ -318,16 +316,14 @@ where
// testing_buildBlockV1: only wire when the hidden testing module is explicitly
// requested on any transport. Default stays disabled to honor security guidance.
let mut testing_api = TestingApi::new(
let testing_api = TestingApi::new(
container.registry.eth_api().clone(),
container.registry.evm_config().clone(),
);
if testing_skip_invalid_transactions {
testing_api = testing_api.with_skip_invalid_transactions();
}
)
.into_rpc();
container
.modules
.merge_if_module_configured(RethRpcModule::Testing, testing_api.into_rpc())?;
.merge_if_module_configured(RethRpcModule::Testing, testing_api)?;
Ok(())
})

View File

@@ -153,7 +153,7 @@ async fn maintain_txpool_reorg() -> eyre::Result<()> {
w1.address(),
);
let pooled_tx1 = EthPooledTransaction::new(tx1.clone(), 200);
let tx_hash1 = *pooled_tx1.hash();
let tx_hash1 = *pooled_tx1.clone().hash();
// build tx2 from wallet2
let envelop2 = TransactionTestContext::transfer_tx(1, w2.clone()).await;
@@ -162,7 +162,7 @@ async fn maintain_txpool_reorg() -> eyre::Result<()> {
w2.address(),
);
let pooled_tx2 = EthPooledTransaction::new(tx2.clone(), 200);
let tx_hash2 = *pooled_tx2.hash();
let tx_hash2 = *pooled_tx2.clone().hash();
let block_info = BlockInfo {
block_gas_limit: ETHEREUM_BLOCK_GAS_LIMIT_30M,

View File

@@ -31,8 +31,8 @@ reth-payload-validator.workspace = true
# ethereum
alloy-rlp.workspace = true
alloy-rpc-types-engine.workspace = true
revm.workspace = true
alloy-rpc-types-engine.workspace = true
# alloy
alloy-eips.workspace = true

View File

@@ -155,7 +155,7 @@ where
let state_provider = client.state_by_block_hash(parent_header.hash())?;
let state = StateProviderDatabase::new(state_provider.as_ref());
let mut db =
State::builder().with_database(cached_reads.as_db_mut(state)).with_bundle_update().build();
State::builder().with_database_ref(cached_reads.as_db(state)).with_bundle_update().build();
let mut builder = evm_config
.builder_for_next_block(
@@ -247,7 +247,7 @@ where
limit: MAX_RLP_BLOCK_SIZE,
},
);
continue
continue;
}
// There's only limited amount of blob space available per block, so we need to check if

View File

@@ -236,7 +236,7 @@ impl reth_codecs::Compact for Transaction {
// # Panics
//
// A panic will be triggered if an identifier larger than 3 is passed from the database. For
// optimism an identifier with value [`DEPOSIT_TX_TYPE_ID`] is allowed.
// optimism a identifier with value [`DEPOSIT_TX_TYPE_ID`] is allowed.
fn from_compact(buf: &[u8], identifier: usize) -> (Self, &[u8]) {
let (tx_type, buf) = TxType::from_compact(buf, identifier);

View File

@@ -741,7 +741,6 @@ mod tests {
nonce,
code_hash: KECCAK_EMPTY,
code: None,
account_id: None,
};
state.insert_account(addr, account_info);
state
@@ -778,13 +777,8 @@ mod tests {
let mut state = setup_state_with_account(addr1, 100, 1);
let account2 = AccountInfo {
balance: U256::from(200),
nonce: 1,
code_hash: KECCAK_EMPTY,
code: None,
account_id: None,
};
let account2 =
AccountInfo { balance: U256::from(200), nonce: 1, code_hash: KECCAK_EMPTY, code: None };
state.insert_account(addr2, account2);
let mut increments = HashMap::default();
@@ -805,13 +799,8 @@ mod tests {
let mut state = setup_state_with_account(addr1, 100, 1);
let account2 = AccountInfo {
balance: U256::from(200),
nonce: 1,
code_hash: KECCAK_EMPTY,
code: None,
account_id: None,
};
let account2 =
AccountInfo { balance: U256::from(200), nonce: 1, code_hash: KECCAK_EMPTY, code: None };
state.insert_account(addr2, account2);
let mut increments = HashMap::default();

View File

@@ -399,7 +399,7 @@ pub trait ConfigureEvm: Clone + Debug + Send + Sync + Unpin {
/// // Complete block building
/// let outcome = builder.finish(state_provider)?;
/// ```
fn builder_for_next_block<'a, DB: Database + 'a>(
fn builder_for_next_block<'a, DB: Database>(
&'a self,
db: &'a mut State<DB>,
parent: &'a SealedHeader<<Self::Primitives as NodePrimitives>::BlockHeader>,

View File

@@ -1,7 +1,7 @@
//! Contains [Chain], a chain of blocks and their final state.
use crate::ExecutionOutcome;
use alloc::{borrow::Cow, collections::BTreeMap, vec::Vec};
use alloc::{borrow::Cow, collections::BTreeMap, sync::Arc, vec::Vec};
use alloy_consensus::{transaction::Recovered, BlockHeader};
use alloy_eips::{eip1898::ForkBlock, eip2718::Encodable2718, BlockNumHash};
use alloy_primitives::{Address, BlockHash, BlockNumber, TxHash};
@@ -10,7 +10,7 @@ use reth_primitives_traits::{
transaction::signed::SignedTransaction, Block, BlockBody, IndexedTx, NodePrimitives,
RecoveredBlock, SealedHeader,
};
use reth_trie_common::LazyTrieData;
use reth_trie_common::{updates::TrieUpdatesSorted, HashedPostStateSorted};
/// A chain of blocks and their final state.
///
@@ -34,10 +34,10 @@ pub struct Chain<N: NodePrimitives = reth_ethereum_primitives::EthPrimitives> {
///
/// Additionally, it includes the individual state changes that led to the current state.
execution_outcome: ExecutionOutcome<N::Receipt>,
/// Lazy trie data for each block in the chain, keyed by block number.
///
/// Contains handles to lazily-initialized sorted trie updates and hashed state.
trie_data: BTreeMap<BlockNumber, LazyTrieData>,
/// State trie updates for each block in the chain, keyed by block number.
trie_updates: BTreeMap<BlockNumber, Arc<TrieUpdatesSorted>>,
/// Hashed post state for each block in the chain, keyed by block number.
hashed_state: BTreeMap<BlockNumber, Arc<HashedPostStateSorted>>,
}
type ChainTxReceiptMeta<'a, N> = (
@@ -52,7 +52,8 @@ impl<N: NodePrimitives> Default for Chain<N> {
Self {
blocks: Default::default(),
execution_outcome: Default::default(),
trie_data: Default::default(),
trie_updates: Default::default(),
hashed_state: Default::default(),
}
}
}
@@ -66,23 +67,27 @@ impl<N: NodePrimitives> Chain<N> {
pub fn new(
blocks: impl IntoIterator<Item = RecoveredBlock<N::Block>>,
execution_outcome: ExecutionOutcome<N::Receipt>,
trie_data: BTreeMap<BlockNumber, LazyTrieData>,
trie_updates: BTreeMap<BlockNumber, Arc<TrieUpdatesSorted>>,
hashed_state: BTreeMap<BlockNumber, Arc<HashedPostStateSorted>>,
) -> Self {
let blocks =
blocks.into_iter().map(|b| (b.header().number(), b)).collect::<BTreeMap<_, _>>();
debug_assert!(!blocks.is_empty(), "Chain should have at least one block");
Self { blocks, execution_outcome, trie_data }
Self { blocks, execution_outcome, trie_updates, hashed_state }
}
/// Create new Chain from a single block and its state.
pub fn from_block(
block: RecoveredBlock<N::Block>,
execution_outcome: ExecutionOutcome<N::Receipt>,
trie_data: LazyTrieData,
trie_updates: Arc<TrieUpdatesSorted>,
hashed_state: Arc<HashedPostStateSorted>,
) -> Self {
let block_number = block.header().number();
Self::new([block], execution_outcome, BTreeMap::from([(block_number, trie_data)]))
let trie_updates_map = BTreeMap::from([(block_number, trie_updates)]);
let hashed_state_map = BTreeMap::from([(block_number, hashed_state)]);
Self::new([block], execution_outcome, trie_updates_map, hashed_state_map)
}
/// Get the blocks in this chain.
@@ -100,19 +105,37 @@ impl<N: NodePrimitives> Chain<N> {
self.blocks.values().map(|block| block.clone_sealed_header())
}
/// Get all trie data for this chain.
pub const fn trie_data(&self) -> &BTreeMap<BlockNumber, LazyTrieData> {
&self.trie_data
/// Get all trie updates for this chain.
pub const fn trie_updates(&self) -> &BTreeMap<BlockNumber, Arc<TrieUpdatesSorted>> {
&self.trie_updates
}
/// Get trie data for a specific block number.
pub fn trie_data_at(&self, block_number: BlockNumber) -> Option<&LazyTrieData> {
self.trie_data.get(&block_number)
/// Get trie updates for a specific block number.
pub fn trie_updates_at(&self, block_number: BlockNumber) -> Option<&Arc<TrieUpdatesSorted>> {
self.trie_updates.get(&block_number)
}
/// Remove all trie data for this chain.
pub fn clear_trie_data(&mut self) {
self.trie_data.clear();
/// Remove all trie updates for this chain.
pub fn clear_trie_updates(&mut self) {
self.trie_updates.clear();
}
/// Get all hashed states for this chain.
pub const fn hashed_state(&self) -> &BTreeMap<BlockNumber, Arc<HashedPostStateSorted>> {
&self.hashed_state
}
/// Get hashed state for a specific block number.
pub fn hashed_state_at(
&self,
block_number: BlockNumber,
) -> Option<&Arc<HashedPostStateSorted>> {
self.hashed_state.get(&block_number)
}
/// Remove all hashed states for this chain.
pub fn clear_hashed_state(&mut self) {
self.hashed_state.clear();
}
/// Get execution outcome of this chain
@@ -160,16 +183,23 @@ impl<N: NodePrimitives> Chain<N> {
/// Destructure the chain into its inner components:
/// 1. The blocks contained in the chain.
/// 2. The execution outcome representing the final state.
/// 3. The trie data map.
/// 3. The trie updates map.
/// 4. The hashed state map.
#[allow(clippy::type_complexity)]
pub fn into_inner(
self,
) -> (
ChainBlocks<'static, N::Block>,
ExecutionOutcome<N::Receipt>,
BTreeMap<BlockNumber, LazyTrieData>,
BTreeMap<BlockNumber, Arc<TrieUpdatesSorted>>,
BTreeMap<BlockNumber, Arc<HashedPostStateSorted>>,
) {
(ChainBlocks { blocks: Cow::Owned(self.blocks) }, self.execution_outcome, self.trie_data)
(
ChainBlocks { blocks: Cow::Owned(self.blocks) },
self.execution_outcome,
self.trie_updates,
self.hashed_state,
)
}
/// Destructure the chain into its inner components:
@@ -299,12 +329,14 @@ impl<N: NodePrimitives> Chain<N> {
&mut self,
block: RecoveredBlock<N::Block>,
execution_outcome: ExecutionOutcome<N::Receipt>,
trie_data: LazyTrieData,
trie_updates: Arc<TrieUpdatesSorted>,
hashed_state: Arc<HashedPostStateSorted>,
) {
let block_number = block.header().number();
self.blocks.insert(block_number, block);
self.execution_outcome.extend(execution_outcome);
self.trie_data.insert(block_number, trie_data);
self.trie_updates.insert(block_number, trie_updates);
self.hashed_state.insert(block_number, hashed_state);
}
/// Merge two chains by appending the given chain into the current one.
@@ -323,7 +355,8 @@ impl<N: NodePrimitives> Chain<N> {
// Insert blocks from other chain
self.blocks.extend(other.blocks);
self.execution_outcome.extend(other.execution_outcome);
self.trie_data.extend(other.trie_data);
self.trie_updates.extend(other.trie_updates);
self.hashed_state.extend(other.hashed_state);
Ok(())
}
@@ -550,14 +583,14 @@ pub(super) mod serde_bincode_compat {
execution_outcome: value.execution_outcome.as_repr(),
_trie_updates_legacy: None,
trie_updates: value
.trie_data
.trie_updates
.iter()
.map(|(k, v)| (*k, v.get().trie_updates.as_ref().into()))
.map(|(k, v)| (*k, v.as_ref().into()))
.collect(),
hashed_state: value
.trie_data
.hashed_state
.iter()
.map(|(k, v)| (*k, v.get().hashed_state.as_ref().into()))
.map(|(k, v)| (*k, v.as_ref().into()))
.collect(),
}
}
@@ -570,24 +603,19 @@ pub(super) mod serde_bincode_compat {
>,
{
fn from(value: Chain<'a, N>) -> Self {
use reth_trie_common::LazyTrieData;
let hashed_state_map: BTreeMap<_, _> =
value.hashed_state.into_iter().map(|(k, v)| (k, Arc::new(v.into()))).collect();
let trie_data: BTreeMap<BlockNumber, LazyTrieData> = value
.trie_updates
.into_iter()
.map(|(k, v)| {
let hashed_state = hashed_state_map.get(&k).cloned().unwrap_or_default();
(k, LazyTrieData::ready(hashed_state, Arc::new(v.into())))
})
.collect();
Self {
blocks: value.blocks.0.into_owned(),
execution_outcome: ExecutionOutcome::from_repr(value.execution_outcome),
trie_data,
trie_updates: value
.trie_updates
.into_iter()
.map(|(k, v)| (k, Arc::new(v.into())))
.collect(),
hashed_state: value
.hashed_state
.into_iter()
.map(|(k, v)| (k, Arc::new(v.into())))
.collect(),
}
}
}
@@ -648,6 +676,7 @@ pub(super) mod serde_bincode_compat {
.unwrap()],
Default::default(),
BTreeMap::new(),
BTreeMap::new(),
),
};
@@ -747,8 +776,12 @@ mod tests {
let mut block_state_extended = execution_outcome1;
block_state_extended.extend(execution_outcome2);
let chain: Chain =
Chain::new(vec![block1.clone(), block2.clone()], block_state_extended, BTreeMap::new());
let chain: Chain = Chain::new(
vec![block1.clone(), block2.clone()],
block_state_extended,
BTreeMap::new(),
BTreeMap::new(),
);
// return tip state
assert_eq!(

View File

@@ -1,5 +1,3 @@
use alloy_primitives::{Address, B256, U256};
use reth_primitives_traits::{Account, Bytecode};
use revm::database::BundleState;
pub use alloy_evm::block::BlockExecutionResult;
@@ -25,36 +23,3 @@ pub struct BlockExecutionOutput<T> {
/// The changed state of the block after execution.
pub state: BundleState,
}
impl<T> BlockExecutionOutput<T> {
/// Return bytecode if known.
pub fn bytecode(&self, code_hash: &B256) -> Option<Bytecode> {
self.state.bytecode(code_hash).map(Bytecode)
}
/// Get account if account is known.
pub fn account(&self, address: &Address) -> Option<Option<Account>> {
self.state.account(address).map(|a| a.info.as_ref().map(Into::into))
}
/// Get storage if value is known.
///
/// This means that depending on status we can potentially return `U256::ZERO`.
pub fn storage(&self, address: &Address, storage_key: U256) -> Option<U256> {
self.state.account(address).and_then(|a| a.storage_slot(storage_key))
}
}
impl<T> Default for BlockExecutionOutput<T> {
fn default() -> Self {
Self {
result: BlockExecutionResult {
receipts: Default::default(),
requests: Default::default(),
gas_used: 0,
blob_gas_used: 0,
},
state: Default::default(),
}
}
}

View File

@@ -249,14 +249,6 @@ impl<T> ExecutionOutcome<T> {
&self.receipts[index]
}
/// Returns an iterator over receipt slices, one per block.
///
/// This is a more ergonomic alternative to `receipts()` that yields slices
/// instead of requiring indexing into a nested `Vec<Vec<T>>`.
pub fn receipts_iter(&self) -> impl Iterator<Item = &[T]> + '_ {
self.receipts.iter().map(|v| v.as_slice())
}
/// Is execution outcome empty.
pub const fn is_empty(&self) -> bool {
self.len() == 0
@@ -942,20 +934,10 @@ mod tests {
let address3 = Address::random();
// Set up account info with some changes
let account_info1 = AccountInfo {
nonce: 1,
balance: U256::from(100),
code_hash: B256::ZERO,
code: None,
account_id: None,
};
let account_info2 = AccountInfo {
nonce: 2,
balance: U256::from(200),
code_hash: B256::ZERO,
code: None,
account_id: None,
};
let account_info1 =
AccountInfo { nonce: 1, balance: U256::from(100), code_hash: B256::ZERO, code: None };
let account_info2 =
AccountInfo { nonce: 2, balance: U256::from(200), code_hash: B256::ZERO, code: None };
// Set up the bundle state with these accounts
let mut bundle_state = BundleState::default();

View File

@@ -149,7 +149,7 @@ where
executor.into_state().take_bundle(),
results,
);
let chain = Chain::new(blocks, outcome, BTreeMap::new());
let chain = Chain::new(blocks, outcome, BTreeMap::new(), BTreeMap::new());
Ok(chain)
}
}

View File

@@ -503,7 +503,6 @@ where
}
break
}
let buffer_full = this.buffer.len() >= this.max_capacity;
// Update capacity
this.update_capacity();
@@ -537,12 +536,6 @@ where
// Update capacity
this.update_capacity();
// If the buffer was full and we made space, we need to wake up to accept new notifications
if buffer_full && this.buffer.len() < this.max_capacity {
debug!(target: "exex::manager", "Buffer has space again, waking up senders");
cx.waker().wake_by_ref();
}
// Update watch channel block number
let finished_height = this.exex_handles.iter_mut().try_fold(u64::MAX, |curr, exex| {
exex.finished_height.map_or(Err(()), |height| Ok(height.number.min(curr)))
@@ -694,6 +687,7 @@ mod tests {
BlockWriter, Chain, DBProvider, DatabaseProviderFactory, TransactionVariant,
};
use reth_testing_utils::generators::{self, random_block, BlockParams};
use std::collections::BTreeMap;
fn empty_finalized_header_stream() -> ForkChoiceStream<SealedHeader> {
let (tx, rx) = watch::channel(None);
@@ -795,7 +789,12 @@ mod tests {
block1.set_block_number(10);
let notification1 = ExExNotification::ChainCommitted {
new: Arc::new(Chain::new(vec![block1.clone()], Default::default(), Default::default())),
new: Arc::new(Chain::new(
vec![block1.clone()],
Default::default(),
Default::default(),
Default::default(),
)),
};
// Push the first notification
@@ -813,7 +812,12 @@ mod tests {
block2.set_block_number(20);
let notification2 = ExExNotification::ChainCommitted {
new: Arc::new(Chain::new(vec![block2.clone()], Default::default(), Default::default())),
new: Arc::new(Chain::new(
vec![block2.clone()],
Default::default(),
Default::default(),
Default::default(),
)),
};
exex_manager.push_notification(notification2.clone());
@@ -856,7 +860,12 @@ mod tests {
block1.set_block_number(10);
let notification1 = ExExNotification::ChainCommitted {
new: Arc::new(Chain::new(vec![block1.clone()], Default::default(), Default::default())),
new: Arc::new(Chain::new(
vec![block1.clone()],
Default::default(),
Default::default(),
Default::default(),
)),
};
exex_manager.push_notification(notification1.clone());
@@ -1084,6 +1093,7 @@ mod tests {
vec![Default::default()],
Default::default(),
Default::default(),
Default::default(),
)),
};
@@ -1154,6 +1164,7 @@ mod tests {
vec![Default::default()],
Default::default(),
Default::default(),
Default::default(),
)),
};
@@ -1198,7 +1209,12 @@ mod tests {
block1.set_block_number(10);
let notification = ExExNotification::ChainCommitted {
new: Arc::new(Chain::new(vec![block1.clone()], Default::default(), Default::default())),
new: Arc::new(Chain::new(
vec![block1.clone()],
Default::default(),
Default::default(),
Default::default(),
)),
};
let mut cx = Context::from_waker(futures::task::noop_waker_ref());
@@ -1347,11 +1363,17 @@ mod tests {
new: Arc::new(Chain::new(
vec![genesis_block.clone()],
Default::default(),
Default::default(),
BTreeMap::new(),
BTreeMap::new(),
)),
};
let notification = ExExNotification::ChainCommitted {
new: Arc::new(Chain::new(vec![block.clone()], Default::default(), Default::default())),
new: Arc::new(Chain::new(
vec![block.clone()],
Default::default(),
BTreeMap::new(),
BTreeMap::new(),
)),
};
let (finalized_headers_tx, rx) = watch::channel(None);
@@ -1421,78 +1443,4 @@ mod tests {
Ok(())
}
#[tokio::test]
async fn test_deadlock_manager_wakes_after_buffer_clears() {
// This test simulates the scenario where the buffer fills up, ingestion pauses,
// and then space clears. We verify the manager wakes up to process pending items.
let temp_dir = tempfile::tempdir().unwrap();
let wal = Wal::new(temp_dir.path()).unwrap();
let provider_factory = create_test_provider_factory();
init_genesis(&provider_factory).unwrap();
let provider = BlockchainProvider::new(provider_factory.clone()).unwrap();
// 1. Setup Manager with Capacity = 1
let (exex_handle, _, mut notifications) = ExExHandle::new(
"test_exex".to_string(),
Default::default(),
provider,
EthEvmConfig::mainnet(),
wal.handle(),
);
let max_capacity = 2;
let exex_manager = ExExManager::new(
provider_factory,
vec![exex_handle],
max_capacity,
wal,
empty_finalized_header_stream(),
);
let manager_handle = exex_manager.handle();
// Spawn manager in background so it runs continuously
tokio::spawn(async move {
exex_manager.await.ok();
});
// Helper to create notifications
let mut rng = generators::rng();
let mut make_notif = |id: u64| {
let block = random_block(&mut rng, id, BlockParams::default()).try_recover().unwrap();
ExExNotification::ChainCommitted {
new: Arc::new(Chain::new(vec![block], Default::default(), Default::default())),
}
};
manager_handle.send(ExExNotificationSource::Pipeline, make_notif(1)).unwrap();
// Send the "Stuck" Item (Notification #100).
// At this point, the Manager loop has skipped the ingestion logic because buffer is full
// (buffer_full=true). This item sits in the unbounded 'handle_rx' channel waiting.
manager_handle.send(ExExNotificationSource::Pipeline, make_notif(100)).unwrap();
// 3. Relieve Pressure
// We consume items from the ExEx.
// As we pull items out, the ExEx frees space -> Manager sends buffered item -> Manager
// frees space. Once Manager frees space, the FIX (wake_by_ref) should trigger,
// causing it to read Notif #100.
// Consume the jam
let _ = notifications.next().await.unwrap();
// 4. Assert No Deadlock
// We expect Notification #100 next.
// If the wake_by_ref fix is missing, this will Time Out because the manager is sleeping
// despite having empty buffer.
let result =
tokio::time::timeout(std::time::Duration::from_secs(1), notifications.next()).await;
assert!(
result.is_ok(),
"Deadlock detected! Manager failed to wake up and process Pending Item #100."
);
}
}

View File

@@ -501,6 +501,7 @@ mod tests {
.try_recover()?],
Default::default(),
BTreeMap::new(),
BTreeMap::new(),
)),
};
@@ -569,6 +570,7 @@ mod tests {
.try_recover()?],
Default::default(),
BTreeMap::new(),
BTreeMap::new(),
)),
};
@@ -636,6 +638,7 @@ mod tests {
vec![exex_head_block.clone().try_recover()?],
Default::default(),
BTreeMap::new(),
BTreeMap::new(),
)),
};
wal.commit(&exex_head_notification)?;
@@ -650,6 +653,7 @@ mod tests {
.try_recover()?],
Default::default(),
BTreeMap::new(),
BTreeMap::new(),
)),
};
@@ -707,6 +711,7 @@ mod tests {
vec![exex_head_block.clone().try_recover()?],
Default::default(),
BTreeMap::new(),
BTreeMap::new(),
)),
};
wal.commit(&exex_head_notification)?;
@@ -726,6 +731,7 @@ mod tests {
.try_recover()?],
Default::default(),
BTreeMap::new(),
BTreeMap::new(),
)),
};

View File

@@ -304,24 +304,37 @@ mod tests {
vec![blocks[0].clone(), blocks[1].clone()],
Default::default(),
BTreeMap::new(),
BTreeMap::new(),
)),
};
let reverted_notification = ExExNotification::ChainReverted {
old: Arc::new(Chain::new(vec![blocks[1].clone()], Default::default(), BTreeMap::new())),
old: Arc::new(Chain::new(
vec![blocks[1].clone()],
Default::default(),
BTreeMap::new(),
BTreeMap::new(),
)),
};
let committed_notification_2 = ExExNotification::ChainCommitted {
new: Arc::new(Chain::new(
vec![block_1_reorged.clone(), blocks[2].clone()],
Default::default(),
BTreeMap::new(),
BTreeMap::new(),
)),
};
let reorged_notification = ExExNotification::ChainReorged {
old: Arc::new(Chain::new(vec![blocks[2].clone()], Default::default(), BTreeMap::new())),
old: Arc::new(Chain::new(
vec![blocks[2].clone()],
Default::default(),
BTreeMap::new(),
BTreeMap::new(),
)),
new: Arc::new(Chain::new(
vec![block_2_reorged.clone(), blocks[3].clone()],
Default::default(),
BTreeMap::new(),
BTreeMap::new(),
)),
};

View File

@@ -189,7 +189,7 @@ mod tests {
use reth_testing_utils::generators::{self, random_block};
use reth_trie_common::{
updates::{StorageTrieUpdates, TrieUpdates},
BranchNodeCompact, HashedPostState, HashedStorage, LazyTrieData, Nibbles,
BranchNodeCompact, HashedPostState, HashedStorage, Nibbles,
};
use std::{collections::BTreeMap, fs::File, sync::Arc};
@@ -241,8 +241,18 @@ mod tests {
let new_block = random_block(&mut rng, 0, Default::default()).try_recover()?;
let notification = ExExNotification::ChainReorged {
new: Arc::new(Chain::new(vec![new_block], Default::default(), BTreeMap::new())),
old: Arc::new(Chain::new(vec![old_block], Default::default(), BTreeMap::new())),
new: Arc::new(Chain::new(
vec![new_block],
Default::default(),
BTreeMap::new(),
BTreeMap::new(),
)),
old: Arc::new(Chain::new(
vec![old_block],
Default::default(),
BTreeMap::new(),
BTreeMap::new(),
)),
};
// Do a round trip serialization and deserialization
@@ -336,17 +346,13 @@ mod tests {
)]),
};
let trie_data = LazyTrieData::ready(
Arc::new(hashed_state.into_sorted()),
Arc::new(trie_updates.into_sorted()),
);
let notification: ExExNotification<reth_ethereum_primitives::EthPrimitives> =
ExExNotification::ChainCommitted {
new: Arc::new(Chain::new(
vec![block],
Default::default(),
BTreeMap::from([(block_number, trie_data)]),
BTreeMap::from([(block_number, Arc::new(trie_updates.into_sorted()))]),
BTreeMap::from([(block_number, Arc::new(hashed_state.into_sorted()))]),
)),
};
Ok(notification)

View File

@@ -247,7 +247,7 @@ pub async fn test_exex_context_with_chain_spec(
db,
chain_spec.clone(),
StaticFileProvider::read_write(static_dir.keep()).expect("static file provider"),
RocksDBProvider::builder(rocksdb_dir.keep()).build().unwrap(),
RocksDBProvider::builder(rocksdb_dir.keep()).with_default_tables().build().unwrap(),
)?;
let genesis_hash = init_genesis(&provider_factory)?;

View File

@@ -223,12 +223,14 @@ pub(super) mod serde_bincode_compat {
.unwrap()],
Default::default(),
BTreeMap::new(),
BTreeMap::new(),
)),
new: Arc::new(Chain::new(
vec![RecoveredBlock::arbitrary(&mut arbitrary::Unstructured::new(&bytes))
.unwrap()],
Default::default(),
BTreeMap::new(),
BTreeMap::new(),
)),
},
};

View File

@@ -106,7 +106,7 @@ impl BanList {
self.banned_ips.contains_key(ip)
}
/// checks the ban list to see if it contains the given peer
/// checks the ban list to see if it contains the given ip
#[inline]
pub fn is_banned_peer(&self, peer_id: &PeerId) -> bool {
self.banned_peers.contains_key(peer_id)
@@ -117,7 +117,7 @@ impl BanList {
self.banned_ips.remove(ip);
}
/// Unbans the peer
/// Unbans the ip address
pub fn unban_peer(&mut self, peer_id: &PeerId) {
self.banned_peers.remove(peer_id);
}

View File

@@ -54,7 +54,6 @@ reth-tasks.workspace = true
reth-tokio-util.workspace = true
reth-tracing.workspace = true
reth-transaction-pool.workspace = true
reth-trie-db = { workspace = true, features = ["metrics"] }
reth-basic-payload-builder.workspace = true
reth-node-ethstats.workspace = true
@@ -116,7 +115,6 @@ test-utils = [
"reth-db-api/test-utils",
"reth-provider/test-utils",
"reth-transaction-pool/test-utils",
"reth-trie-db/test-utils",
"reth-evm-ethereum/test-utils",
"reth-node-ethereum/test-utils",
"reth-primitives-traits/test-utils",

View File

@@ -84,7 +84,6 @@ use reth_tracing::{
tracing::{debug, error, info, warn},
};
use reth_transaction_pool::TransactionPool;
use reth_trie_db::ChangesetCache;
use std::{sync::Arc, thread::available_parallelism, time::Duration};
use tokio::sync::{
mpsc::{unbounded_channel, UnboundedSender},
@@ -471,10 +470,7 @@ where
/// Returns the [`ProviderFactory`] for the attached storage after executing a consistent check
/// between the database and static files. **It may execute a pipeline unwind if it fails this
/// check.**
pub async fn create_provider_factory<N, Evm>(
&self,
changeset_cache: ChangesetCache,
) -> eyre::Result<ProviderFactory<N>>
pub async fn create_provider_factory<N, Evm>(&self) -> eyre::Result<ProviderFactory<N>>
where
N: ProviderNodeTypes<DB = DB, ChainSpec = ChainSpec>,
Evm: ConfigureEvm<Primitives = N::Primitives> + 'static,
@@ -504,8 +500,7 @@ where
static_file_provider,
rocksdb_provider,
)?
.with_prune_modes(self.prune_modes())
.with_changeset_cache(changeset_cache);
.with_prune_modes(self.prune_modes());
// Keep MDBX, static files, and RocksDB aligned. If any check fails, unwind to the
// earliest consistent block.
@@ -598,13 +593,12 @@ where
/// Creates a new [`ProviderFactory`] and attaches it to the launch context.
pub async fn with_provider_factory<N, Evm>(
self,
changeset_cache: ChangesetCache,
) -> eyre::Result<LaunchContextWith<Attached<WithConfigs<ChainSpec>, ProviderFactory<N>>>>
where
N: ProviderNodeTypes<DB = DB, ChainSpec = ChainSpec>,
Evm: ConfigureEvm<Primitives = N::Primitives> + 'static,
{
let factory = self.create_provider_factory::<N, Evm>(changeset_cache).await?;
let factory = self.create_provider_factory::<N, Evm>().await?;
let ctx = LaunchContextWith {
inner: self.inner,
attachment: self.attachment.map_right(|_| factory),

View File

@@ -37,7 +37,6 @@ use reth_provider::{
use reth_tasks::TaskExecutor;
use reth_tokio_util::EventSender;
use reth_tracing::tracing::{debug, error, info};
use reth_trie_db::ChangesetCache;
use std::{future::Future, pin::Pin, sync::Arc};
use tokio::sync::{mpsc::unbounded_channel, oneshot};
use tokio_stream::wrappers::UnboundedReceiverStream;
@@ -88,9 +87,6 @@ impl EngineNodeLauncher {
} = target;
let NodeHooks { on_component_initialized, on_node_started, .. } = hooks;
// Create changeset cache that will be shared across the engine
let changeset_cache = ChangesetCache::new();
// setup the launch context
let ctx = ctx
.with_configured_globals(engine_tree_config.reserved_cpu_cores())
@@ -102,8 +98,8 @@ impl EngineNodeLauncher {
.attach(database.clone())
// ensure certain settings take effect
.with_adjusted_configs()
// Create the provider factory with changeset cache
.with_provider_factory::<_, <CB::Components as NodeComponents<T>>::Evm>(changeset_cache.clone()).await?
// Create the provider factory
.with_provider_factory::<_, <CB::Components as NodeComponents<T>>::Evm>().await?
.inspect(|ctx| {
info!(target: "reth::cli", "Database opened");
match ctx.provider_factory().storage_settings() {
@@ -208,7 +204,7 @@ impl EngineNodeLauncher {
// Build the engine validator with all required components
let engine_validator = validator_builder
.clone()
.build_tree_validator(&add_ons_ctx, engine_tree_config.clone(), changeset_cache.clone())
.build_tree_validator(&add_ons_ctx, engine_tree_config.clone())
.await?;
// Create the consensus engine stream with optional reorg
@@ -218,13 +214,7 @@ impl EngineNodeLauncher {
.maybe_reorg(
ctx.blockchain_db().clone(),
ctx.components().evm_config().clone(),
|| async {
// Create a separate cache for reorg validator (not shared with main engine)
let reorg_cache = ChangesetCache::new();
validator_builder
.build_tree_validator(&add_ons_ctx, engine_tree_config.clone(), reorg_cache)
.await
},
|| validator_builder.build_tree_validator(&add_ons_ctx, engine_tree_config.clone()),
node_config.debug.reorg_frequency,
node_config.debug.reorg_depth,
)
@@ -249,7 +239,6 @@ impl EngineNodeLauncher {
engine_tree_config,
ctx.sync_metrics_tx(),
ctx.components().evm_config().clone(),
changeset_cache,
);
info!(target: "reth::cli", "Consensus engine initialized");

View File

@@ -3,7 +3,6 @@
pub use jsonrpsee::server::middleware::rpc::{RpcService, RpcServiceBuilder};
pub use reth_engine_tree::tree::{BasicEngineValidator, EngineValidator};
pub use reth_rpc_builder::{middleware::RethRpcMiddleware, Identity, Stack};
pub use reth_trie_db::ChangesetCache;
use crate::{
invalid_block_hook::InvalidBlockHookExt, ConfigureEngineEvm, ConsensusEngineEvent,
@@ -1289,7 +1288,6 @@ pub trait EngineValidatorBuilder<Node: FullNodeComponents>: Send + Sync + Clone
self,
ctx: &AddOnsContext<'_, Node>,
tree_config: TreeConfig,
changeset_cache: ChangesetCache,
) -> impl Future<Output = eyre::Result<Self::EngineValidator>> + Send;
}
@@ -1337,12 +1335,10 @@ where
self,
ctx: &AddOnsContext<'_, Node>,
tree_config: TreeConfig,
changeset_cache: ChangesetCache,
) -> eyre::Result<Self::EngineValidator> {
let validator = self.payload_validator_builder.build(ctx).await?;
let data_dir = ctx.config.datadir.clone().resolve_datadir(ctx.config.chain.chain());
let invalid_block_hook = ctx.create_invalid_block_hook(&data_dir).await?;
Ok(BasicEngineValidator::new(
ctx.node.provider().clone(),
std::sync::Arc::new(ctx.node.consensus().clone()),
@@ -1350,7 +1346,6 @@ where
validator,
tree_config,
invalid_block_hook,
changeset_cache,
))
}
}

View File

@@ -81,8 +81,7 @@ tokio.workspace = true
jemalloc = ["reth-cli-util/jemalloc"]
asm-keccak = ["alloy-primitives/asm-keccak"]
keccak-cache-global = ["alloy-primitives/keccak-cache-global"]
otlp = ["reth-tracing/otlp", "reth-tracing-otlp/otlp"]
otlp-logs = ["reth-tracing/otlp-logs", "reth-tracing-otlp/otlp-logs"]
otlp = ["reth-tracing/otlp"]
tracy = ["reth-tracing/tracy"]
min-error-logs = ["tracing/release_max_level_error"]

View File

@@ -26,7 +26,7 @@ pub use log::{ColorMode, LogArgs, Verbosity};
/// `TraceArgs` for tracing and spans support
mod trace;
pub use trace::{OtlpInitStatus, OtlpLogsStatus, TraceArgs};
pub use trace::{OtlpInitStatus, TraceArgs};
/// `MetricArgs` to configure metrics.
mod metric;

View File

@@ -5,7 +5,10 @@ use alloy_primitives::{Address, BlockNumber};
use clap::{builder::RangedU64ValueParser, Args};
use reth_chainspec::EthereumHardforks;
use reth_config::config::PruneConfig;
use reth_prune_types::{PruneMode, PruneModes, ReceiptsLogPruneConfig, MINIMUM_PRUNING_DISTANCE};
use reth_prune_types::{
PruneMode, PruneModes, ReceiptsLogPruneConfig, MERKLE_CHANGESETS_RETENTION_BLOCKS,
MINIMUM_PRUNING_DISTANCE,
};
use std::{collections::BTreeMap, ops::Not};
/// Parameters for pruning and full node
@@ -140,6 +143,7 @@ impl PruningArgs {
.ethereum_fork_activation(EthereumHardfork::Paris)
.block_number()
.map(PruneMode::Before),
merkle_changesets: PruneMode::Distance(MERKLE_CHANGESETS_RETENTION_BLOCKS),
receipts_log_filter: Default::default(),
},
}
@@ -156,6 +160,7 @@ impl PruningArgs {
account_history: Some(PruneMode::Distance(10064)),
storage_history: Some(PruneMode::Distance(10064)),
bodies_history: Some(PruneMode::Distance(10064)),
merkle_changesets: PruneMode::Distance(MERKLE_CHANGESETS_RETENTION_BLOCKS),
receipts_log_filter: Default::default(),
},
}

View File

@@ -640,13 +640,6 @@ pub struct RpcServerArgs {
value_parser = parse_duration_from_secs_or_ms,
)]
pub rpc_send_raw_transaction_sync_timeout: Duration,
/// Skip invalid transactions in `testing_buildBlockV1` instead of failing.
///
/// When enabled, transactions that fail execution will be skipped, and all subsequent
/// transactions from the same sender will also be skipped.
#[arg(long = "testing.skip-invalid-transactions", default_value_t = false)]
pub testing_skip_invalid_transactions: bool,
}
impl RpcServerArgs {
@@ -859,7 +852,6 @@ impl Default for RpcServerArgs {
rpc_state_cache,
gas_price_oracle,
rpc_send_raw_transaction_sync_timeout,
testing_skip_invalid_transactions: false,
}
}
}
@@ -1034,7 +1026,6 @@ mod tests {
default_suggested_fee: None,
},
rpc_send_raw_transaction_sync_timeout: std::time::Duration::from_secs(30),
testing_skip_invalid_transactions: true,
};
let parsed_args = CommandParser::<RpcServerArgs>::parse_from([
@@ -1123,7 +1114,6 @@ mod tests {
"60",
"--rpc.send-raw-transaction-sync-timeout",
"30s",
"--testing.skip-invalid-transactions",
])
.args;

View File

@@ -38,6 +38,11 @@ pub enum StageEnum {
///
/// Handles Merkle tree-related computations and data processing.
Merkle,
/// The merkle changesets stage within the pipeline.
///
/// Handles Merkle trie changesets for storage and accounts.
#[value(name = "merkle-changesets")]
MerkleChangeSets,
/// The transaction lookup stage within the pipeline.
///
/// Deals with the retrieval and processing of transactions.

View File

@@ -61,6 +61,16 @@ pub struct StaticFilesArgs {
/// the node has been initialized, changing this flag requires re-syncing from scratch.
#[arg(long = "static-files.account-change-sets")]
pub account_changesets: bool,
/// Use `RocksDB` for history indices instead of MDBX.
///
/// When enabled, `AccountsHistory`, `StoragesHistory`, and `TransactionHashNumbers`
/// tables will be stored in `RocksDB` for better write performance.
///
/// Note: This setting can only be configured at genesis initialization. Once
/// the node has been initialized, changing this flag requires re-syncing from scratch.
#[arg(long = "storage.rocksdb")]
pub rocksdb: bool,
}
impl StaticFilesArgs {
@@ -101,5 +111,8 @@ impl StaticFilesArgs {
.with_receipts_in_static_files(self.receipts)
.with_transaction_senders_in_static_files(self.transaction_senders)
.with_account_changesets_in_static_files(self.account_changesets)
.with_account_history_in_rocksdb(self.rocksdb)
.with_storages_history_in_rocksdb(self.rocksdb)
.with_transaction_hash_numbers_in_rocksdb(self.rocksdb)
}
}

View File

@@ -1,4 +1,4 @@
//! Opentelemetry tracing and logging configuration through CLI args.
//! Opentelemetry tracing configuration through CLI args.
use clap::Parser;
use eyre::WrapErr;
@@ -6,7 +6,7 @@ use reth_tracing::{tracing_subscriber::EnvFilter, Layers};
use reth_tracing_otlp::OtlpProtocol;
use url::Url;
/// CLI arguments for configuring `Opentelemetry` trace and logs export.
/// CLI arguments for configuring `Opentelemetry` trace and span export.
#[derive(Debug, Clone, Parser)]
pub struct TraceArgs {
/// Enable `Opentelemetry` tracing export to an OTLP endpoint.
@@ -30,29 +30,9 @@ pub struct TraceArgs {
)]
pub otlp: Option<Url>,
/// Enable `Opentelemetry` logs export to an OTLP endpoint.
/// OTLP transport protocol to use for exporting traces.
///
/// If no value provided, defaults based on protocol:
/// - HTTP: `http://localhost:4318/v1/logs`
/// - gRPC: `http://localhost:4317`
///
/// Example: --logs-otlp=http://collector:4318/v1/logs
#[arg(
long = "logs-otlp",
env = "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT",
global = true,
value_name = "URL",
num_args = 0..=1,
default_missing_value = "http://localhost:4318/v1/logs",
require_equals = true,
value_parser = parse_otlp_endpoint,
help_heading = "Logging"
)]
pub logs_otlp: Option<Url>,
/// OTLP transport protocol to use for exporting traces and logs.
///
/// - `http`: expects endpoint path to end with `/v1/traces` or `/v1/logs`
/// - `http`: expects endpoint path to end with `/v1/traces`
/// - `grpc`: expects endpoint without a path
///
/// Defaults to HTTP if not specified.
@@ -82,22 +62,6 @@ pub struct TraceArgs {
)]
pub otlp_filter: EnvFilter,
/// Set a filter directive for the OTLP logs exporter. This controls the verbosity
/// of logs sent to the OTLP endpoint. It follows the same syntax as the
/// `RUST_LOG` environment variable.
///
/// Example: --logs-otlp.filter=info,reth=debug
///
/// Defaults to INFO if not specified.
#[arg(
long = "logs-otlp.filter",
global = true,
value_name = "FILTER",
default_value = "info",
help_heading = "Logging"
)]
pub logs_otlp_filter: EnvFilter,
/// Service name to use for OTLP tracing export.
///
/// This name will be used to identify the service in distributed tracing systems
@@ -137,10 +101,8 @@ impl Default for TraceArgs {
fn default() -> Self {
Self {
otlp: None,
logs_otlp: None,
protocol: OtlpProtocol::Http,
otlp_filter: EnvFilter::from_default_env(),
logs_otlp_filter: EnvFilter::try_new("info").expect("valid filter"),
sample_ratio: None,
service_name: "reth".to_string(),
}
@@ -188,37 +150,6 @@ impl TraceArgs {
Ok(OtlpInitStatus::Disabled)
}
}
/// Initialize OTLP logs export with the given layers.
///
/// This method handles OTLP logs initialization based on the configured options,
/// including validation and protocol selection.
///
/// Returns the initialization status to allow callers to log appropriate messages.
pub async fn init_otlp_logs(&mut self, _layers: &mut Layers) -> eyre::Result<OtlpLogsStatus> {
if let Some(endpoint) = self.logs_otlp.as_mut() {
self.protocol.validate_logs_endpoint(endpoint)?;
#[cfg(feature = "otlp-logs")]
{
let config = reth_tracing_otlp::OtlpLogsConfig::new(
self.service_name.clone(),
endpoint.clone(),
self.protocol,
)?;
_layers.with_log_layer(config.clone(), self.logs_otlp_filter.clone())?;
Ok(OtlpLogsStatus::Started(config.endpoint().clone()))
}
#[cfg(not(feature = "otlp-logs"))]
{
Ok(OtlpLogsStatus::NoFeature)
}
} else {
Ok(OtlpLogsStatus::Disabled)
}
}
}
/// Status of OTLP tracing initialization.
@@ -232,17 +163,6 @@ pub enum OtlpInitStatus {
NoFeature,
}
/// Status of OTLP logs initialization.
#[derive(Debug)]
pub enum OtlpLogsStatus {
/// OTLP logs export was successfully started with the given endpoint.
Started(Url),
/// OTLP logs export is disabled (no endpoint configured).
Disabled,
/// OTLP logs arguments provided but feature is not compiled.
NoFeature,
}
// Parses an OTLP endpoint url.
fn parse_otlp_endpoint(arg: &str) -> eyre::Result<Url> {
Url::parse(arg).wrap_err("Invalid URL for OTLP trace output")

View File

@@ -342,6 +342,14 @@ impl<ChainSpec> NodeConfig<ChainSpec> {
self
}
/// Converts the node configuration to [`StorageSettings`].
///
/// This returns storage settings configured via CLI arguments including
/// static file settings and `RocksDB` settings.
pub const fn to_storage_settings(&self) -> reth_provider::StorageSettings {
self.static_files.to_settings()
}
/// Returns pruning configuration.
pub fn prune_config(&self) -> Option<PruneConfig>
where

View File

@@ -43,7 +43,6 @@ tracy = ["reth-optimism-cli/tracy"]
asm-keccak = ["reth-optimism-cli/asm-keccak", "reth-optimism-node/asm-keccak"]
keccak-cache-global = [
"reth-optimism-cli/keccak-cache-global",
"reth-optimism-node/keccak-cache-global",
]
dev = [

View File

@@ -76,9 +76,8 @@ reth-optimism-chainspec = { workspace = true, features = ["std", "superchain-con
[features]
default = []
# Opentelemetry feature to activate tracing and logs export
# Opentelemtry feature to activate metrics export
otlp = ["reth-tracing/otlp", "reth-node-core/otlp"]
otlp-logs = ["reth-tracing/otlp-logs", "reth-node-core/otlp-logs"]
asm-keccak = [
"alloy-primitives/asm-keccak",
@@ -86,12 +85,6 @@ asm-keccak = [
"reth-optimism-node/asm-keccak",
]
keccak-cache-global = [
"alloy-primitives/keccak-cache-global",
"reth-node-core/keccak-cache-global",
"reth-optimism-node/keccak-cache-global",
]
# Jemalloc feature for vergen to generate correct env vars
jemalloc = [
"reth-node-core/jemalloc",

View File

@@ -3,7 +3,7 @@ use eyre::{eyre, Result};
use reth_cli::chainspec::ChainSpecParser;
use reth_cli_commands::launcher::Launcher;
use reth_cli_runner::CliRunner;
use reth_node_core::args::{OtlpInitStatus, OtlpLogsStatus};
use reth_node_core::args::OtlpInitStatus;
use reth_node_metrics::recorder::install_prometheus_recorder;
use reth_optimism_chainspec::OpChainSpec;
use reth_optimism_consensus::OpBeaconConsensus;
@@ -124,11 +124,9 @@ where
let mut layers = self.layers.take().unwrap_or_default();
let otlp_status = runner.block_on(self.cli.traces.init_otlp_tracing(&mut layers))?;
let otlp_logs_status = runner.block_on(self.cli.traces.init_otlp_logs(&mut layers))?;
self.guard = self.cli.logs.init_tracing_with_layers(layers)?;
info!(target: "reth::cli", "Initialized tracing, debug log directory: {}", self.cli.logs.log_file_directory);
match otlp_status {
OtlpInitStatus::Started(endpoint) => {
info!(target: "reth::cli", "Started OTLP {:?} tracing export to {endpoint}", self.cli.traces.protocol);
@@ -138,16 +136,6 @@ where
}
OtlpInitStatus::Disabled => {}
}
match otlp_logs_status {
OtlpLogsStatus::Started(endpoint) => {
info!(target: "reth::cli", "Started OTLP {:?} logs export to {endpoint}", self.cli.traces.protocol);
}
OtlpLogsStatus::NoFeature => {
warn!(target: "reth::cli", "Provided OTLP logs arguments do not have effect, compile with the `otlp-logs` feature")
}
OtlpLogsStatus::Disabled => {}
}
}
Ok(())
}

View File

@@ -18,7 +18,7 @@ use alloy_consensus::{
use alloy_primitives::B64;
use core::fmt::Debug;
use reth_chainspec::EthChainSpec;
use reth_consensus::{Consensus, ConsensusError, FullConsensus, HeaderValidator, ReceiptRootBloom};
use reth_consensus::{Consensus, ConsensusError, FullConsensus, HeaderValidator};
use reth_consensus_common::validation::{
validate_against_parent_eip1559_base_fee, validate_against_parent_hash_number,
validate_against_parent_timestamp, validate_cancun_gas, validate_header_base_fee,
@@ -79,9 +79,8 @@ where
&self,
block: &RecoveredBlock<N::Block>,
result: &BlockExecutionResult<N::Receipt>,
receipt_root_bloom: Option<ReceiptRootBloom>,
) -> Result<(), ConsensusError> {
validate_block_post_execution(block.header(), &self.chain_spec, result, receipt_root_bloom)
validate_block_post_execution(block.header(), &self.chain_spec, result)
}
}
@@ -411,8 +410,7 @@ mod tests {
let post_execution = <OpBeaconConsensus<OpChainSpec> as FullConsensus<OpPrimitives>>::validate_block_post_execution(
&beacon_consensus,
&block,
&result,
None,
&result
);
// validate blob, it should pass blob gas used validation
@@ -481,8 +479,7 @@ mod tests {
let post_execution = <OpBeaconConsensus<OpChainSpec> as FullConsensus<OpPrimitives>>::validate_block_post_execution(
&beacon_consensus,
&block,
&result,
None,
&result
);
// validate blob, it should fail blob gas used validation post execution.

View File

@@ -85,14 +85,10 @@ where
///
/// - Compares the receipts root in the block header to the block body
/// - Compares the gas used in the block header to the actual gas usage after execution
///
/// If `receipt_root_bloom` is provided, the pre-computed receipt root and logs bloom are used
/// instead of computing them from the receipts.
pub fn validate_block_post_execution<R: DepositReceipt>(
header: impl BlockHeader,
chain_spec: impl OpHardforks,
result: &BlockExecutionResult<R>,
receipt_root_bloom: Option<(B256, Bloom)>,
) -> Result<(), ConsensusError> {
// Validate that the blob gas used is present and correctly computed if Jovian is active.
if chain_spec.is_jovian_active_at_timestamp(header.timestamp()) {
@@ -114,32 +110,21 @@ pub fn validate_block_post_execution<R: DepositReceipt>(
// operation as hashing that is required for state root got calculated in every
// transaction This was replaced with is_success flag.
// See more about EIP here: https://eips.ethereum.org/EIPS/eip-658
if chain_spec.is_byzantium_active_at_block(header.number()) {
let result = if let Some((receipts_root, logs_bloom)) = receipt_root_bloom {
compare_receipts_root_and_logs_bloom(
receipts_root,
logs_bloom,
header.receipts_root(),
header.logs_bloom(),
)
} else {
verify_receipts_optimism(
header.receipts_root(),
header.logs_bloom(),
receipts,
chain_spec,
header.timestamp(),
)
};
if let Err(error) = result {
let receipts = receipts
.iter()
.map(|r| Bytes::from(r.with_bloom_ref().encoded_2718()))
.collect::<Vec<_>>();
tracing::debug!(%error, ?receipts, "receipts verification failed");
return Err(error)
}
if chain_spec.is_byzantium_active_at_block(header.number()) &&
let Err(error) = verify_receipts_optimism(
header.receipts_root(),
header.logs_bloom(),
receipts,
chain_spec,
header.timestamp(),
)
{
let receipts = receipts
.iter()
.map(|r| Bytes::from(r.with_bloom_ref().encoded_2718()))
.collect::<Vec<_>>();
tracing::debug!(%error, ?receipts, "receipts verification failed");
return Err(error)
}
// Check if gas used matches the value set in header.
@@ -558,7 +543,7 @@ mod tests {
requests: Requests::default(),
gas_used: GAS_USED,
};
validate_block_post_execution(&header, &chainspec, &result, None).unwrap();
validate_block_post_execution(&header, &chainspec, &result).unwrap();
}
#[test]
@@ -580,7 +565,7 @@ mod tests {
gas_used: GAS_USED,
};
assert!(matches!(
validate_block_post_execution(&header, &chainspec, &result, None).unwrap_err(),
validate_block_post_execution(&header, &chainspec, &result).unwrap_err(),
ConsensusError::BlobGasUsedDiff(diff)
if diff.got == BLOB_GAS_USED && diff.expected == BLOB_GAS_USED + 1
));

View File

@@ -230,9 +230,7 @@ where
let spec = revm_spec_by_timestamp_after_bedrock(self.chain_spec(), timestamp);
let cfg_env = CfgEnv::new()
.with_chain_id(self.chain_spec().chain().id())
.with_spec_and_mainnet_gas_params(spec);
let cfg_env = CfgEnv::new().with_chain_id(self.chain_spec().chain().id()).with_spec(spec);
let blob_excess_gas_and_price = spec
.into_eth_spec()
@@ -364,8 +362,7 @@ mod tests {
let db = CacheDB::<EmptyDBTyped<ProviderError>>::default();
// Create a custom configuration environment with a chain ID of 111
let cfg =
CfgEnv::new().with_chain_id(111).with_spec_and_mainnet_gas_params(OpSpecId::default());
let cfg = CfgEnv::new().with_chain_id(111).with_spec(OpSpecId::default());
let evm_env = EvmEnv { cfg_env: cfg.clone(), ..Default::default() };
@@ -403,10 +400,8 @@ mod tests {
let db = CacheDB::<EmptyDBTyped<ProviderError>>::default();
let evm_env = EvmEnv {
cfg_env: CfgEnv::new().with_spec_and_mainnet_gas_params(OpSpecId::ECOTONE),
..Default::default()
};
let evm_env =
EvmEnv { cfg_env: CfgEnv::new().with_spec(OpSpecId::ECOTONE), ..Default::default() };
let evm = evm_config.evm_with_env(db, evm_env.clone());
@@ -432,8 +427,7 @@ mod tests {
let evm_config = test_evm_config();
let db = CacheDB::<EmptyDBTyped<ProviderError>>::default();
let cfg =
CfgEnv::new().with_chain_id(111).with_spec_and_mainnet_gas_params(OpSpecId::default());
let cfg = CfgEnv::new().with_chain_id(111).with_spec(OpSpecId::default());
let block = BlockEnv::default();
let evm_env = EvmEnv { block_env: block, cfg_env: cfg.clone() };
@@ -469,10 +463,8 @@ mod tests {
let evm_config = test_evm_config();
let db = CacheDB::<EmptyDBTyped<ProviderError>>::default();
let evm_env = EvmEnv {
cfg_env: CfgEnv::new().with_spec_and_mainnet_gas_params(OpSpecId::ECOTONE),
..Default::default()
};
let evm_env =
EvmEnv { cfg_env: CfgEnv::new().with_spec(OpSpecId::ECOTONE), ..Default::default() };
let evm = evm_config.evm_with_env_and_inspector(db, evm_env.clone(), NoOpInspector {});
@@ -529,8 +521,12 @@ mod tests {
// Create a Chain object with a BTreeMap of blocks mapped to their block numbers,
// including block1_hash and block2_hash, and the execution_outcome
let chain: Chain<OpPrimitives> =
Chain::new([block1, block2], execution_outcome.clone(), BTreeMap::new());
let chain: Chain<OpPrimitives> = Chain::new(
[block1, block2],
execution_outcome.clone(),
BTreeMap::new(),
BTreeMap::new(),
);
// Assert that the proper receipt vector is returned for block1_hash
assert_eq!(chain.receipts_by_block_hash(block1_hash), Some(vec![&receipt1]));

View File

@@ -12,18 +12,10 @@ use reth_primitives_traits::{AlloyBlockHeader, BlockTy, HeaderTy, NodePrimitives
use reth_revm::cached::CachedReads;
use reth_storage_api::{BlockReaderIdExt, StateProviderFactory};
use reth_tasks::TaskExecutor;
use std::{
sync::Arc,
time::{Duration, Instant},
};
use tokio::{
sync::{oneshot, watch},
time::sleep,
};
use std::{sync::Arc, time::Instant};
use tokio::sync::{oneshot, watch};
use tracing::*;
const CONNECTION_BACKOUT_PERIOD: Duration = Duration::from_secs(5);
/// The `FlashBlockService` maintains an in-memory [`PendingFlashBlock`] built out of a sequence of
/// [`FlashBlock`]s.
#[derive(Debug)]
@@ -175,13 +167,7 @@ where
self.try_start_build_job();
}
Some(Err(err)) => {
warn!(
target: "flashblocks",
%err,
retry_period = CONNECTION_BACKOUT_PERIOD.as_secs(),
"Error receiving flashblock"
);
sleep(CONNECTION_BACKOUT_PERIOD).await;
warn!(target: "flashblocks", %err, "Error receiving flashblock");
}
None => {
warn!(target: "flashblocks", "Flashblock stream ended");

Some files were not shown because too many files have changed in this diff Show More