mirror of
https://github.com/paradigmxyz/reth.git
synced 2026-04-30 03:01:58 -04:00
Compare commits
12 Commits
eth/70
...
performanc
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
5c04d1abe1 | ||
|
|
4231f4b688 | ||
|
|
0b607113dc | ||
|
|
be4dc53b92 | ||
|
|
4afb555d06 | ||
|
|
ab2ef99458 | ||
|
|
bfd4b79245 | ||
|
|
49057b1c0c | ||
|
|
b6772370d7 | ||
|
|
d72935628a | ||
|
|
ad63b135d6 | ||
|
|
90651ae8e8 |
@@ -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,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",
|
||||
"--features=std,op,dev,asm-keccak,jemalloc,jemalloc-prof,tracy-allocator,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.
|
||||
|
||||
43
Cargo.lock
generated
43
Cargo.lock
generated
@@ -329,9 +329,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "alloy-json-abi"
|
||||
version = "1.4.1"
|
||||
version = "1.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5513d5e6bd1cba6bdcf5373470f559f320c05c8c59493b6e98912fbe6733943f"
|
||||
checksum = "6bfca3dbbcb7498f0f60e67aff2ad6aff57032e22eb2fd03189854be11a22c03"
|
||||
dependencies = [
|
||||
"alloy-primitives",
|
||||
"alloy-sol-type-parser",
|
||||
@@ -426,9 +426,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "alloy-primitives"
|
||||
version = "1.4.1"
|
||||
version = "1.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "355bf68a433e0fd7f7d33d5a9fc2583fde70bf5c530f63b80845f8da5505cf28"
|
||||
checksum = "5c850e6ccbd34b8a463a1e934ffc8fc00e1efc5e5489f2ad82d7797949f3bd4e"
|
||||
dependencies = [
|
||||
"alloy-rlp",
|
||||
"arbitrary",
|
||||
@@ -447,6 +447,7 @@ dependencies = [
|
||||
"proptest",
|
||||
"proptest-derive 0.6.0",
|
||||
"rand 0.9.2",
|
||||
"rapidhash",
|
||||
"ruint",
|
||||
"rustc-hash",
|
||||
"serde",
|
||||
@@ -781,9 +782,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "alloy-sol-macro"
|
||||
version = "1.4.1"
|
||||
version = "1.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f3ce480400051b5217f19d6e9a82d9010cdde20f1ae9c00d53591e4a1afbb312"
|
||||
checksum = "b2218e3aeb3ee665d117fdf188db0d5acfdc3f7b7502c827421cb78f26a2aec0"
|
||||
dependencies = [
|
||||
"alloy-sol-macro-expander",
|
||||
"alloy-sol-macro-input",
|
||||
@@ -795,9 +796,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "alloy-sol-macro-expander"
|
||||
version = "1.4.1"
|
||||
version = "1.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6d792e205ed3b72f795a8044c52877d2e6b6e9b1d13f431478121d8d4eaa9028"
|
||||
checksum = "b231cb8cc48e66dd1c6e11a1402f3ac86c3667cbc13a6969a0ac030ba7bb8c88"
|
||||
dependencies = [
|
||||
"alloy-sol-macro-input",
|
||||
"const-hex",
|
||||
@@ -813,9 +814,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "alloy-sol-macro-input"
|
||||
version = "1.4.1"
|
||||
version = "1.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0bd1247a8f90b465ef3f1207627547ec16940c35597875cdc09c49d58b19693c"
|
||||
checksum = "49a522d79929c1bf0152b07567a38f7eaed3ab149e53e7528afa78ff11994668"
|
||||
dependencies = [
|
||||
"const-hex",
|
||||
"dunce",
|
||||
@@ -829,9 +830,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "alloy-sol-type-parser"
|
||||
version = "1.4.1"
|
||||
version = "1.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "954d1b2533b9b2c7959652df3076954ecb1122a28cc740aa84e7b0a49f6ac0a9"
|
||||
checksum = "0475c459859c8d9428af6ff3736614655a57efda8cc435a3b8b4796fa5ac1dd0"
|
||||
dependencies = [
|
||||
"serde",
|
||||
"winnow",
|
||||
@@ -839,9 +840,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "alloy-sol-types"
|
||||
version = "1.4.1"
|
||||
version = "1.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "70319350969a3af119da6fb3e9bddb1bce66c9ea933600cb297c8b1850ad2a3c"
|
||||
checksum = "35287d9d821d5f26011bcd8d9101340898f761c9933cf50fca689bb7ed62fdeb"
|
||||
dependencies = [
|
||||
"alloy-json-abi",
|
||||
"alloy-primitives",
|
||||
@@ -7209,6 +7210,16 @@ dependencies = [
|
||||
"rand_core 0.9.3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rapidhash"
|
||||
version = "4.1.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d8e65c75143ce5d47c55b510297eeb1182f3c739b6043c537670e9fc18612dae"
|
||||
dependencies = [
|
||||
"rand 0.9.2",
|
||||
"rustversion",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ratatui"
|
||||
version = "0.29.0"
|
||||
@@ -12340,9 +12351,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "syn-solidity"
|
||||
version = "1.4.1"
|
||||
version = "1.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ff790eb176cc81bb8936aed0f7b9f14fc4670069a2d371b3e3b0ecce908b2cb3"
|
||||
checksum = "60ceeb7c95a4536de0c0e1649bd98d1a72a4bb9590b1f3e45a8a0bfdb7c188c0"
|
||||
dependencies = [
|
||||
"paste",
|
||||
"proc-macro2",
|
||||
|
||||
@@ -489,10 +489,10 @@ alloy-dyn-abi = "1.4.1"
|
||||
alloy-eip2124 = { version = "0.2.0", default-features = false }
|
||||
alloy-eip7928 = { version = "0.1.0" }
|
||||
alloy-evm = { version = "0.25.1", default-features = false }
|
||||
alloy-primitives = { version = "1.4.1", default-features = false, features = ["map-foldhash"] }
|
||||
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.4.1"
|
||||
alloy-sol-types = { version = "1.4.1", default-features = false }
|
||||
alloy-sol-macro = "1.5.0"
|
||||
alloy-sol-types = { version = "1.5.0", default-features = false }
|
||||
alloy-trie = { version = "0.9.1", default-features = false }
|
||||
|
||||
alloy-hardforks = "0.4.5"
|
||||
|
||||
@@ -81,7 +81,7 @@ backon.workspace = true
|
||||
tempfile.workspace = true
|
||||
|
||||
[features]
|
||||
default = ["jemalloc", "otlp", "reth-revm/portable", "js-tracer"]
|
||||
default = ["jemalloc", "otlp", "reth-revm/portable", "js-tracer", "keccak-cache-global"]
|
||||
|
||||
otlp = [
|
||||
"reth-ethereum-cli/otlp",
|
||||
@@ -102,7 +102,9 @@ asm-keccak = [
|
||||
"reth-ethereum-cli/asm-keccak",
|
||||
"reth-node-ethereum/asm-keccak",
|
||||
]
|
||||
|
||||
keccak-cache-global = [
|
||||
"reth-node-ethereum/keccak-cache-global",
|
||||
]
|
||||
jemalloc = [
|
||||
"reth-cli-util/jemalloc",
|
||||
"reth-node-core/jemalloc",
|
||||
|
||||
@@ -112,6 +112,7 @@ impl<C: ChainSpecParser> EnvironmentArgs<C> {
|
||||
};
|
||||
// TransactionDB only support read-write mode
|
||||
let rocksdb_provider = RocksDBProvider::builder(data_dir.rocksdb())
|
||||
.with_default_tables()
|
||||
.with_database_log_level(self.db.log_level)
|
||||
.build()?;
|
||||
|
||||
|
||||
@@ -241,6 +241,7 @@ fn bench_state_root(c: &mut Criterion) {
|
||||
StateProviderBuilder::new(provider.clone(), genesis_hash, None),
|
||||
OverlayStateProviderFactory::new(provider),
|
||||
&TreeConfig::default(),
|
||||
None,
|
||||
);
|
||||
|
||||
let mut state_hook = handle.state_hook();
|
||||
|
||||
318
crates/engine/tree/src/tree/payload_processor/bal.rs
Normal file
318
crates/engine/tree/src/tree/payload_processor/bal.rs
Normal file
@@ -0,0 +1,318 @@
|
||||
//! BAL (Block Access List, EIP-7928) related functionality.
|
||||
|
||||
use alloy_consensus::constants::KECCAK_EMPTY;
|
||||
use alloy_eip7928::BlockAccessList;
|
||||
use alloy_primitives::{keccak256, U256};
|
||||
use reth_primitives_traits::Account;
|
||||
use reth_provider::{AccountReader, ProviderError};
|
||||
use reth_trie::{HashedPostState, HashedStorage};
|
||||
|
||||
/// Converts a Block Access List into a [`HashedPostState`] by extracting the final state
|
||||
/// of modified accounts and storage slots.
|
||||
pub fn bal_to_hashed_post_state<P>(
|
||||
bal: &BlockAccessList,
|
||||
provider: &P,
|
||||
) -> Result<HashedPostState, ProviderError>
|
||||
where
|
||||
P: AccountReader,
|
||||
{
|
||||
let mut hashed_state = HashedPostState::with_capacity(bal.len());
|
||||
|
||||
for account_changes in bal {
|
||||
let address = account_changes.address;
|
||||
let hashed_address = keccak256(address);
|
||||
|
||||
// Get the latest balance (last balance change if any)
|
||||
let balance = account_changes.balance_changes.last().map(|change| change.post_balance);
|
||||
|
||||
// Get the latest nonce (last nonce change if any)
|
||||
let nonce = account_changes.nonce_changes.last().map(|change| change.new_nonce);
|
||||
|
||||
// Get the latest code (last code change if any)
|
||||
let code_hash = if let Some(code_change) = account_changes.code_changes.last() {
|
||||
if code_change.new_code.is_empty() {
|
||||
Some(Some(KECCAK_EMPTY))
|
||||
} else {
|
||||
Some(Some(keccak256(&code_change.new_code)))
|
||||
}
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
// Only fetch account from provider if we're missing any field
|
||||
let existing_account = if balance.is_none() || nonce.is_none() || code_hash.is_none() {
|
||||
provider.basic_account(&address)?
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
// Build the final account state
|
||||
let account = Account {
|
||||
balance: balance.unwrap_or_else(|| {
|
||||
existing_account.as_ref().map(|acc| acc.balance).unwrap_or(U256::ZERO)
|
||||
}),
|
||||
nonce: nonce
|
||||
.unwrap_or_else(|| existing_account.as_ref().map(|acc| acc.nonce).unwrap_or(0)),
|
||||
bytecode_hash: code_hash.unwrap_or_else(|| {
|
||||
existing_account.as_ref().and_then(|acc| acc.bytecode_hash).or(Some(KECCAK_EMPTY))
|
||||
}),
|
||||
};
|
||||
|
||||
hashed_state.accounts.insert(hashed_address, Some(account));
|
||||
|
||||
// Process storage changes
|
||||
if !account_changes.storage_changes.is_empty() {
|
||||
let mut storage_map = HashedStorage::new(false);
|
||||
|
||||
for slot_changes in &account_changes.storage_changes {
|
||||
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, U256::from_be_bytes(last_change.new_value.0));
|
||||
}
|
||||
}
|
||||
|
||||
if !storage_map.storage.is_empty() {
|
||||
hashed_state.storages.insert(hashed_address, storage_map);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(hashed_state)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use alloy_eip7928::{
|
||||
AccountChanges, BalanceChange, CodeChange, NonceChange, SlotChanges, StorageChange,
|
||||
};
|
||||
use alloy_primitives::{Address, Bytes, StorageKey, B256};
|
||||
use reth_revm::test_utils::StateProviderTest;
|
||||
|
||||
#[test]
|
||||
fn test_bal_to_hashed_post_state_basic() {
|
||||
let provider = StateProviderTest::default();
|
||||
|
||||
let address = Address::random();
|
||||
let account_changes = AccountChanges {
|
||||
address,
|
||||
storage_changes: vec![],
|
||||
storage_reads: vec![],
|
||||
balance_changes: vec![BalanceChange::new(0, U256::from(100))],
|
||||
nonce_changes: vec![NonceChange::new(0, 1)],
|
||||
code_changes: vec![],
|
||||
};
|
||||
|
||||
let bal = vec![account_changes];
|
||||
let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
|
||||
|
||||
assert_eq!(result.accounts.len(), 1);
|
||||
|
||||
let hashed_address = keccak256(address);
|
||||
let account_opt = result.accounts.get(&hashed_address).unwrap();
|
||||
assert!(account_opt.is_some());
|
||||
|
||||
let account = account_opt.as_ref().unwrap();
|
||||
assert_eq!(account.balance, U256::from(100));
|
||||
assert_eq!(account.nonce, 1);
|
||||
assert_eq!(account.bytecode_hash, Some(KECCAK_EMPTY));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_bal_with_storage_changes() {
|
||||
let provider = StateProviderTest::default();
|
||||
|
||||
let address = Address::random();
|
||||
let slot = StorageKey::random();
|
||||
let value = B256::random();
|
||||
|
||||
let slot_changes = SlotChanges { slot, changes: vec![StorageChange::new(0, value)] };
|
||||
|
||||
let account_changes = AccountChanges {
|
||||
address,
|
||||
storage_changes: vec![slot_changes],
|
||||
storage_reads: vec![],
|
||||
balance_changes: vec![BalanceChange::new(0, U256::from(500))],
|
||||
nonce_changes: vec![NonceChange::new(0, 2)],
|
||||
code_changes: vec![],
|
||||
};
|
||||
|
||||
let bal = vec![account_changes];
|
||||
let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
|
||||
|
||||
let hashed_address = keccak256(address);
|
||||
assert!(result.storages.contains_key(&hashed_address));
|
||||
|
||||
let storage = result.storages.get(&hashed_address).unwrap();
|
||||
let hashed_slot = keccak256(slot);
|
||||
|
||||
let stored_value = storage.storage.get(&hashed_slot).unwrap();
|
||||
assert_eq!(*stored_value, U256::from_be_bytes(value.0));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_bal_with_code_change() {
|
||||
let provider = StateProviderTest::default();
|
||||
|
||||
let address = Address::random();
|
||||
let code = Bytes::from(vec![0x60, 0x80, 0x60, 0x40]); // Some bytecode
|
||||
|
||||
let account_changes = AccountChanges {
|
||||
address,
|
||||
storage_changes: vec![],
|
||||
storage_reads: vec![],
|
||||
balance_changes: vec![BalanceChange::new(0, U256::from(1000))],
|
||||
nonce_changes: vec![NonceChange::new(0, 1)],
|
||||
code_changes: vec![CodeChange::new(0, code.clone())],
|
||||
};
|
||||
|
||||
let bal = vec![account_changes];
|
||||
let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
|
||||
|
||||
let hashed_address = keccak256(address);
|
||||
let account_opt = result.accounts.get(&hashed_address).unwrap();
|
||||
let account = account_opt.as_ref().unwrap();
|
||||
|
||||
let expected_code_hash = keccak256(&code);
|
||||
assert_eq!(account.bytecode_hash, Some(expected_code_hash));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_bal_with_empty_code() {
|
||||
let provider = StateProviderTest::default();
|
||||
|
||||
let address = Address::random();
|
||||
let empty_code = Bytes::default();
|
||||
|
||||
let account_changes = AccountChanges {
|
||||
address,
|
||||
storage_changes: vec![],
|
||||
storage_reads: vec![],
|
||||
balance_changes: vec![BalanceChange::new(0, U256::from(1000))],
|
||||
nonce_changes: vec![NonceChange::new(0, 1)],
|
||||
code_changes: vec![CodeChange::new(0, empty_code)],
|
||||
};
|
||||
|
||||
let bal = vec![account_changes];
|
||||
let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
|
||||
|
||||
let hashed_address = keccak256(address);
|
||||
let account_opt = result.accounts.get(&hashed_address).unwrap();
|
||||
let account = account_opt.as_ref().unwrap();
|
||||
|
||||
assert_eq!(account.bytecode_hash, Some(KECCAK_EMPTY));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_bal_multiple_changes_takes_last() {
|
||||
let provider = StateProviderTest::default();
|
||||
|
||||
let address = Address::random();
|
||||
|
||||
// Multiple balance changes - should take the last one
|
||||
let account_changes = AccountChanges {
|
||||
address,
|
||||
storage_changes: vec![],
|
||||
storage_reads: vec![],
|
||||
balance_changes: vec![
|
||||
BalanceChange::new(0, U256::from(100)),
|
||||
BalanceChange::new(1, U256::from(200)),
|
||||
BalanceChange::new(2, U256::from(300)),
|
||||
],
|
||||
nonce_changes: vec![
|
||||
NonceChange::new(0, 1),
|
||||
NonceChange::new(1, 2),
|
||||
NonceChange::new(2, 3),
|
||||
],
|
||||
code_changes: vec![],
|
||||
};
|
||||
|
||||
let bal = vec![account_changes];
|
||||
let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
|
||||
|
||||
let hashed_address = keccak256(address);
|
||||
let account_opt = result.accounts.get(&hashed_address).unwrap();
|
||||
let account = account_opt.as_ref().unwrap();
|
||||
|
||||
// Should have the last values
|
||||
assert_eq!(account.balance, U256::from(300));
|
||||
assert_eq!(account.nonce, 3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_bal_uses_provider_for_missing_fields() {
|
||||
let mut provider = StateProviderTest::default();
|
||||
|
||||
let address = Address::random();
|
||||
let code_hash = B256::random();
|
||||
let existing_account =
|
||||
Account { balance: U256::from(999), nonce: 42, bytecode_hash: Some(code_hash) };
|
||||
provider.insert_account(address, existing_account, None, Default::default());
|
||||
|
||||
// Only change balance, nonce and code should come from provider
|
||||
let account_changes = AccountChanges {
|
||||
address,
|
||||
storage_changes: vec![],
|
||||
storage_reads: vec![],
|
||||
balance_changes: vec![BalanceChange::new(0, U256::from(1500))],
|
||||
nonce_changes: vec![],
|
||||
code_changes: vec![],
|
||||
};
|
||||
|
||||
let bal = vec![account_changes];
|
||||
let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
|
||||
|
||||
let hashed_address = keccak256(address);
|
||||
let account_opt = result.accounts.get(&hashed_address).unwrap();
|
||||
let account = account_opt.as_ref().unwrap();
|
||||
|
||||
// Balance should be updated
|
||||
assert_eq!(account.balance, U256::from(1500));
|
||||
// Nonce and bytecode_hash should come from provider
|
||||
assert_eq!(account.nonce, 42);
|
||||
assert_eq!(account.bytecode_hash, Some(code_hash));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_bal_multiple_storage_changes_per_slot() {
|
||||
let provider = StateProviderTest::default();
|
||||
|
||||
let address = Address::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, 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>())),
|
||||
],
|
||||
};
|
||||
|
||||
let account_changes = AccountChanges {
|
||||
address,
|
||||
storage_changes: vec![slot_changes],
|
||||
storage_reads: vec![],
|
||||
balance_changes: vec![BalanceChange::new(0, U256::from(100))],
|
||||
nonce_changes: vec![NonceChange::new(0, 1)],
|
||||
code_changes: vec![],
|
||||
};
|
||||
|
||||
let bal = vec![account_changes];
|
||||
let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
|
||||
|
||||
let hashed_address = keccak256(address);
|
||||
let storage = result.storages.get(&hashed_address).unwrap();
|
||||
let hashed_slot = keccak256(slot);
|
||||
|
||||
let stored_value = storage.storage.get(&hashed_slot).unwrap();
|
||||
|
||||
// Should have the last value
|
||||
assert_eq!(*stored_value, U256::from(300));
|
||||
}
|
||||
}
|
||||
@@ -13,6 +13,7 @@ use crate::tree::{
|
||||
sparse_trie::SparseTrieTask,
|
||||
StateProviderBuilder, TreeConfig,
|
||||
};
|
||||
use alloy_eip7928::BlockAccessList;
|
||||
use alloy_eips::eip1898::BlockWithParent;
|
||||
use alloy_evm::{block::StateChangeSource, ToTxEnv};
|
||||
use alloy_primitives::B256;
|
||||
@@ -49,8 +50,9 @@ use std::{
|
||||
},
|
||||
time::Instant,
|
||||
};
|
||||
use tracing::{debug, debug_span, instrument, warn, Span};
|
||||
use tracing::{debug, debug_span, error, instrument, warn, Span};
|
||||
|
||||
pub mod bal;
|
||||
mod configured_sparse_trie;
|
||||
pub mod executor;
|
||||
pub mod multiproof;
|
||||
@@ -212,6 +214,7 @@ where
|
||||
provider_builder: StateProviderBuilder<N, P>,
|
||||
multiproof_provider_factory: F,
|
||||
config: &TreeConfig,
|
||||
bal: Option<Arc<BlockAccessList>>,
|
||||
) -> PayloadHandle<WithTxEnv<TxEnvFor<Evm>, I::Tx>, I::Error>
|
||||
where
|
||||
P: BlockReader + StateProviderFactory + StateReader + Clone + 'static,
|
||||
@@ -252,19 +255,45 @@ where
|
||||
// wire the multiproof task to the prewarm task
|
||||
let to_multi_proof = Some(multi_proof_task.state_root_message_sender());
|
||||
|
||||
let prewarm_handle = self.spawn_caching_with(
|
||||
env,
|
||||
prewarm_rx,
|
||||
transaction_count_hint,
|
||||
provider_builder,
|
||||
to_multi_proof.clone(),
|
||||
);
|
||||
// Handle BAL-based optimization if available
|
||||
let prewarm_handle = if let Some(bal) = bal {
|
||||
// When BAL is present, skip spawning prewarm tasks entirely and send BAL to multiproof
|
||||
debug!(target: "engine::tree::payload_processor", "BAL present, skipping prewarm tasks");
|
||||
|
||||
// Send BAL message immediately to MultiProofTask
|
||||
if let Some(ref sender) = to_multi_proof &&
|
||||
let Err(err) = sender.send(MultiProofMessage::BlockAccessList(bal))
|
||||
{
|
||||
// In this case state root validation will simply fail
|
||||
error!(target: "engine::tree::payload_processor", ?err, "Failed to send BAL to MultiProofTask");
|
||||
}
|
||||
|
||||
// Spawn minimal cache-only task without prewarming
|
||||
self.spawn_caching_with(
|
||||
env,
|
||||
prewarm_rx,
|
||||
transaction_count_hint,
|
||||
provider_builder.clone(),
|
||||
None, // Don't send proof targets when BAL is present
|
||||
)
|
||||
} else {
|
||||
// Normal path: spawn with full prewarming
|
||||
self.spawn_caching_with(
|
||||
env,
|
||||
prewarm_rx,
|
||||
transaction_count_hint,
|
||||
provider_builder.clone(),
|
||||
to_multi_proof.clone(),
|
||||
)
|
||||
};
|
||||
|
||||
// spawn multi-proof task
|
||||
let parent_span = span.clone();
|
||||
self.executor.spawn_blocking(move || {
|
||||
let _enter = parent_span.entered();
|
||||
multi_proof_task.run();
|
||||
// Build a state provider for the multiproof task
|
||||
let provider = provider_builder.build().expect("failed to build provider");
|
||||
multi_proof_task.run(provider);
|
||||
});
|
||||
|
||||
// wire the sparse trie to the state root response receiver
|
||||
@@ -595,7 +624,7 @@ impl<Tx, Err> PayloadHandle<Tx, Err> {
|
||||
|
||||
move |source: StateChangeSource, state: &EvmState| {
|
||||
if let Some(sender) = &to_multi_proof {
|
||||
let _ = sender.send(MultiProofMessage::StateUpdate(source, state.clone()));
|
||||
let _ = sender.send(MultiProofMessage::StateUpdate(source.into(), state.clone()));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1062,6 +1091,7 @@ mod tests {
|
||||
StateProviderBuilder::new(provider_factory.clone(), genesis_hash, None),
|
||||
OverlayStateProviderFactory::new(provider_factory),
|
||||
&TreeConfig::default(),
|
||||
None, // No BAL for test
|
||||
);
|
||||
|
||||
let mut state_hook = handle.state_hook();
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
//! Multiproof task related functionality.
|
||||
|
||||
use crate::tree::payload_processor::bal::bal_to_hashed_post_state;
|
||||
use alloy_eip7928::BlockAccessList;
|
||||
use alloy_evm::block::StateChangeSource;
|
||||
use alloy_primitives::{
|
||||
keccak256,
|
||||
@@ -11,6 +13,7 @@ use dashmap::DashMap;
|
||||
use derive_more::derive::Deref;
|
||||
use metrics::{Gauge, Histogram};
|
||||
use reth_metrics::Metrics;
|
||||
use reth_provider::AccountReader;
|
||||
use reth_revm::state::EvmState;
|
||||
use reth_trie::{
|
||||
added_removed_keys::MultiAddedRemovedKeys, DecodedMultiProof, HashedPostState, HashedStorage,
|
||||
@@ -26,6 +29,30 @@ use reth_trie_parallel::{
|
||||
use std::{collections::BTreeMap, mem, ops::DerefMut, sync::Arc, time::Instant};
|
||||
use tracing::{debug, error, instrument, trace};
|
||||
|
||||
/// Source of state changes, either from EVM execution or from a Block Access List.
|
||||
#[derive(Clone, Copy)]
|
||||
pub enum Source {
|
||||
/// State changes from EVM execution.
|
||||
Evm(StateChangeSource),
|
||||
/// State changes from Block Access List (EIP-7928).
|
||||
BlockAccessList,
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for Source {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
Self::Evm(source) => source.fmt(f),
|
||||
Self::BlockAccessList => f.write_str("BlockAccessList"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl From<StateChangeSource> for Source {
|
||||
fn from(source: StateChangeSource) -> Self {
|
||||
Self::Evm(source)
|
||||
}
|
||||
}
|
||||
|
||||
/// Maximum number of targets to batch together for prefetch batching.
|
||||
/// Prefetches are just proof requests (no state merging), so we allow a higher cap than state
|
||||
/// updates
|
||||
@@ -82,7 +109,7 @@ pub(super) enum MultiProofMessage {
|
||||
/// Prefetch proof targets
|
||||
PrefetchProofs(MultiProofTargets),
|
||||
/// New state update from transaction execution with its source
|
||||
StateUpdate(StateChangeSource, EvmState),
|
||||
StateUpdate(Source, EvmState),
|
||||
/// State update that can be applied to the sparse trie without any new proofs.
|
||||
///
|
||||
/// It can be the case when all accounts and storage slots from the state update were already
|
||||
@@ -93,6 +120,11 @@ pub(super) enum MultiProofMessage {
|
||||
/// The state update that was used to calculate the proof
|
||||
state: HashedPostState,
|
||||
},
|
||||
/// Block Access List (EIP-7928; BAL) containing complete state changes for the block.
|
||||
///
|
||||
/// When received, the task generates a single state update from the BAL and processes it.
|
||||
/// No further messages are expected after receiving this variant.
|
||||
BlockAccessList(Arc<BlockAccessList>),
|
||||
/// Signals state update stream end.
|
||||
///
|
||||
/// This is triggered by block execution, indicating that no additional state updates are
|
||||
@@ -280,7 +312,7 @@ impl StorageMultiproofInput {
|
||||
/// Input parameters for dispatching a multiproof calculation.
|
||||
#[derive(Debug)]
|
||||
struct MultiproofInput {
|
||||
source: Option<StateChangeSource>,
|
||||
source: Option<Source>,
|
||||
hashed_state_update: HashedPostState,
|
||||
proof_targets: MultiProofTargets,
|
||||
proof_sequence_number: u64,
|
||||
@@ -883,9 +915,19 @@ impl MultiProofTask {
|
||||
skip(self, update),
|
||||
fields(accounts = update.len(), chunks = 0)
|
||||
)]
|
||||
fn on_state_update(&mut self, source: StateChangeSource, update: EvmState) -> u64 {
|
||||
fn on_state_update(&mut self, source: Source, update: EvmState) -> u64 {
|
||||
let hashed_state_update = evm_state_to_hashed_post_state(update);
|
||||
self.on_hashed_state_update(source, hashed_state_update)
|
||||
}
|
||||
|
||||
/// Processes a hashed state update and dispatches multiproofs as needed.
|
||||
///
|
||||
/// Returns the number of state updates dispatched (both `EmptyProof` and regular multiproofs).
|
||||
fn on_hashed_state_update(
|
||||
&mut self,
|
||||
source: Source,
|
||||
hashed_state_update: HashedPostState,
|
||||
) -> u64 {
|
||||
// Update removed keys based on the state update.
|
||||
self.multi_added_removed_keys.update_with_state(&hashed_state_update);
|
||||
|
||||
@@ -982,12 +1024,16 @@ impl MultiProofTask {
|
||||
/// This preserves ordering without requeuing onto the channel.
|
||||
///
|
||||
/// Returns `true` if done, `false` to continue.
|
||||
fn process_multiproof_message(
|
||||
fn process_multiproof_message<P>(
|
||||
&mut self,
|
||||
msg: MultiProofMessage,
|
||||
ctx: &mut MultiproofBatchCtx,
|
||||
batch_metrics: &mut MultiproofBatchMetrics,
|
||||
) -> bool {
|
||||
provider: &P,
|
||||
) -> bool
|
||||
where
|
||||
P: AccountReader,
|
||||
{
|
||||
match msg {
|
||||
// Prefetch proofs: batch consecutive prefetch requests up to target/message limits
|
||||
MultiProofMessage::PrefetchProofs(targets) => {
|
||||
@@ -1146,6 +1192,56 @@ impl MultiProofTask {
|
||||
|
||||
false
|
||||
}
|
||||
// Process Block Access List (BAL) - complete state changes provided upfront
|
||||
MultiProofMessage::BlockAccessList(bal) => {
|
||||
trace!(target: "engine::tree::payload_processor::multiproof", "processing MultiProofMessage::BAL");
|
||||
|
||||
if ctx.first_update_time.is_none() {
|
||||
self.metrics
|
||||
.first_update_wait_time_histogram
|
||||
.record(ctx.start.elapsed().as_secs_f64());
|
||||
ctx.first_update_time = Some(Instant::now());
|
||||
debug!(target: "engine::tree::payload_processor::multiproof", "Started state root calculation from BAL");
|
||||
}
|
||||
|
||||
// Convert BAL to HashedPostState and process it
|
||||
match bal_to_hashed_post_state(&bal, &provider) {
|
||||
Ok(hashed_state) => {
|
||||
debug!(
|
||||
target: "engine::tree::payload_processor::multiproof",
|
||||
accounts = hashed_state.accounts.len(),
|
||||
storages = hashed_state.storages.len(),
|
||||
"Processing BAL state update"
|
||||
);
|
||||
|
||||
// Use BlockAccessList as source for BAL-derived state updates
|
||||
batch_metrics.state_update_proofs_requested +=
|
||||
self.on_hashed_state_update(Source::BlockAccessList, hashed_state);
|
||||
}
|
||||
Err(err) => {
|
||||
error!(target: "engine::tree::payload_processor::multiproof", ?err, "Failed to convert BAL to hashed state");
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
// Mark updates as finished since BAL provides complete state
|
||||
ctx.updates_finished_time = Some(Instant::now());
|
||||
|
||||
// Check if we're done (might need to wait for proofs to complete)
|
||||
if self.is_done(
|
||||
batch_metrics.proofs_processed,
|
||||
batch_metrics.state_update_proofs_requested,
|
||||
batch_metrics.prefetch_proofs_requested,
|
||||
ctx.updates_finished(),
|
||||
) {
|
||||
debug!(
|
||||
target: "engine::tree::payload_processor::multiproof",
|
||||
"BAL processed and all proofs complete, ending calculation"
|
||||
);
|
||||
return true;
|
||||
}
|
||||
false
|
||||
}
|
||||
// Signal that no more state updates will arrive
|
||||
MultiProofMessage::FinishedStateUpdates => {
|
||||
trace!(target: "engine::tree::payload_processor::multiproof", "processing MultiProofMessage::FinishedStateUpdates");
|
||||
@@ -1238,7 +1334,10 @@ impl MultiProofTask {
|
||||
target = "engine::tree::payload_processor::multiproof",
|
||||
skip_all
|
||||
)]
|
||||
pub(crate) fn run(mut self) {
|
||||
pub(crate) fn run<P>(mut self, provider: P)
|
||||
where
|
||||
P: AccountReader,
|
||||
{
|
||||
let mut ctx = MultiproofBatchCtx::new(Instant::now());
|
||||
let mut batch_metrics = MultiproofBatchMetrics::default();
|
||||
|
||||
@@ -1248,7 +1347,7 @@ impl MultiProofTask {
|
||||
trace!(target: "engine::tree::payload_processor::multiproof", "entering main channel receiving loop");
|
||||
|
||||
if let Some(msg) = ctx.pending_msg.take() {
|
||||
if self.process_multiproof_message(msg, &mut ctx, &mut batch_metrics) {
|
||||
if self.process_multiproof_message(msg, &mut ctx, &mut batch_metrics, &provider) {
|
||||
break 'main;
|
||||
}
|
||||
continue;
|
||||
@@ -1323,7 +1422,7 @@ impl MultiProofTask {
|
||||
}
|
||||
};
|
||||
|
||||
if self.process_multiproof_message(msg, &mut ctx, &mut batch_metrics) {
|
||||
if self.process_multiproof_message(msg, &mut ctx, &mut batch_metrics, &provider) {
|
||||
break 'main;
|
||||
}
|
||||
}
|
||||
@@ -1374,7 +1473,7 @@ struct MultiproofBatchCtx {
|
||||
/// Reusable buffer for accumulating prefetch targets during batching.
|
||||
accumulated_prefetch_targets: Vec<MultiProofTargets>,
|
||||
/// Reusable buffer for accumulating state updates during batching.
|
||||
accumulated_state_updates: Vec<(StateChangeSource, EvmState)>,
|
||||
accumulated_state_updates: Vec<(Source, EvmState)>,
|
||||
}
|
||||
|
||||
impl MultiproofBatchCtx {
|
||||
@@ -1492,34 +1591,44 @@ where
|
||||
/// are safe to merge because they originate from the same logical execution and can be
|
||||
/// coalesced to amortize proof work.
|
||||
fn can_batch_state_update(
|
||||
batch_source: StateChangeSource,
|
||||
batch_source: Source,
|
||||
batch_update: &EvmState,
|
||||
next_source: StateChangeSource,
|
||||
next_source: Source,
|
||||
next_update: &EvmState,
|
||||
) -> bool {
|
||||
if !same_state_change_source(batch_source, next_source) {
|
||||
if !same_source(batch_source, next_source) {
|
||||
return false;
|
||||
}
|
||||
|
||||
match (batch_source, next_source) {
|
||||
(StateChangeSource::PreBlock(_), StateChangeSource::PreBlock(_)) |
|
||||
(StateChangeSource::PostBlock(_), StateChangeSource::PostBlock(_)) => {
|
||||
batch_update == next_update
|
||||
}
|
||||
(
|
||||
Source::Evm(StateChangeSource::PreBlock(_)),
|
||||
Source::Evm(StateChangeSource::PreBlock(_)),
|
||||
) |
|
||||
(
|
||||
Source::Evm(StateChangeSource::PostBlock(_)),
|
||||
Source::Evm(StateChangeSource::PostBlock(_)),
|
||||
) => batch_update == next_update,
|
||||
_ => true,
|
||||
}
|
||||
}
|
||||
|
||||
/// Checks whether two state change sources refer to the same origin.
|
||||
fn same_state_change_source(lhs: StateChangeSource, rhs: StateChangeSource) -> bool {
|
||||
/// Checks whether two sources refer to the same origin.
|
||||
fn same_source(lhs: Source, rhs: Source) -> bool {
|
||||
match (lhs, rhs) {
|
||||
(StateChangeSource::Transaction(a), StateChangeSource::Transaction(b)) => a == b,
|
||||
(StateChangeSource::PreBlock(a), StateChangeSource::PreBlock(b)) => {
|
||||
mem::discriminant(&a) == mem::discriminant(&b)
|
||||
}
|
||||
(StateChangeSource::PostBlock(a), StateChangeSource::PostBlock(b)) => {
|
||||
mem::discriminant(&a) == mem::discriminant(&b)
|
||||
}
|
||||
(
|
||||
Source::Evm(StateChangeSource::Transaction(a)),
|
||||
Source::Evm(StateChangeSource::Transaction(b)),
|
||||
) => a == b,
|
||||
(
|
||||
Source::Evm(StateChangeSource::PreBlock(a)),
|
||||
Source::Evm(StateChangeSource::PreBlock(b)),
|
||||
) => mem::discriminant(&a) == mem::discriminant(&b),
|
||||
(
|
||||
Source::Evm(StateChangeSource::PostBlock(a)),
|
||||
Source::Evm(StateChangeSource::PostBlock(b)),
|
||||
) => mem::discriminant(&a) == mem::discriminant(&b),
|
||||
(Source::BlockAccessList, Source::BlockAccessList) => true,
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
@@ -1539,7 +1648,8 @@ fn estimate_evm_state_targets(state: &EvmState) -> usize {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use alloy_primitives::map::B256Set;
|
||||
use alloy_eip7928::{AccountChanges, BalanceChange};
|
||||
use alloy_primitives::{map::B256Set, Address};
|
||||
use reth_provider::{
|
||||
providers::OverlayStateProviderFactory, test_utils::create_test_provider_factory,
|
||||
BlockReader, DatabaseProviderFactory, PruneCheckpointReader, StageCheckpointReader,
|
||||
@@ -1548,7 +1658,7 @@ mod tests {
|
||||
use reth_trie::MultiProof;
|
||||
use reth_trie_parallel::proof_task::{ProofTaskCtx, ProofWorkerHandle};
|
||||
use revm_primitives::{B256, U256};
|
||||
use std::sync::OnceLock;
|
||||
use std::sync::{Arc, OnceLock};
|
||||
use tokio::runtime::{Handle, Runtime};
|
||||
|
||||
/// Get a handle to the test runtime, creating it if necessary
|
||||
@@ -2109,8 +2219,8 @@ mod tests {
|
||||
let source = StateChangeSource::Transaction(0);
|
||||
|
||||
let tx = task.state_root_message_sender();
|
||||
tx.send(MultiProofMessage::StateUpdate(source, update1.clone())).unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(source, update2.clone())).unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(source.into(), update1.clone())).unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(source.into(), update2.clone())).unwrap();
|
||||
|
||||
let proofs_requested =
|
||||
if let Ok(MultiProofMessage::StateUpdate(_src, update)) = task.rx.recv() {
|
||||
@@ -2129,7 +2239,7 @@ mod tests {
|
||||
assert!(merged_update.contains_key(&addr1));
|
||||
assert!(merged_update.contains_key(&addr2));
|
||||
|
||||
task.on_state_update(source, merged_update)
|
||||
task.on_state_update(source.into(), merged_update)
|
||||
} else {
|
||||
panic!("Expected StateUpdate message");
|
||||
};
|
||||
@@ -2173,20 +2283,20 @@ mod tests {
|
||||
|
||||
// Queue: A1 (immediate dispatch), B1 (batched), A2 (should become pending)
|
||||
let tx = task.state_root_message_sender();
|
||||
tx.send(MultiProofMessage::StateUpdate(source_a, create_state_update(addr_a1, 100)))
|
||||
tx.send(MultiProofMessage::StateUpdate(source_a.into(), create_state_update(addr_a1, 100)))
|
||||
.unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(source_b, create_state_update(addr_b1, 200)))
|
||||
tx.send(MultiProofMessage::StateUpdate(source_b.into(), create_state_update(addr_b1, 200)))
|
||||
.unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(source_a, create_state_update(addr_a2, 300)))
|
||||
tx.send(MultiProofMessage::StateUpdate(source_a.into(), create_state_update(addr_a2, 300)))
|
||||
.unwrap();
|
||||
|
||||
let mut pending_msg: Option<MultiProofMessage> = None;
|
||||
|
||||
if let Ok(MultiProofMessage::StateUpdate(first_source, _)) = task.rx.recv() {
|
||||
assert!(same_state_change_source(first_source, source_a));
|
||||
assert!(same_source(first_source, source_a.into()));
|
||||
|
||||
// Simulate batching loop for remaining messages
|
||||
let mut accumulated_updates: Vec<(StateChangeSource, EvmState)> = Vec::new();
|
||||
let mut accumulated_updates: Vec<(Source, EvmState)> = Vec::new();
|
||||
let mut accumulated_targets = 0usize;
|
||||
|
||||
loop {
|
||||
@@ -2234,7 +2344,7 @@ mod tests {
|
||||
|
||||
assert_eq!(accumulated_updates.len(), 1, "Should only batch matching sources");
|
||||
let batch_source = accumulated_updates[0].0;
|
||||
assert!(same_state_change_source(batch_source, source_b));
|
||||
assert!(same_source(batch_source, source_b.into()));
|
||||
|
||||
let batch_source = accumulated_updates[0].0;
|
||||
let mut merged_update = accumulated_updates.remove(0).1;
|
||||
@@ -2242,10 +2352,7 @@ mod tests {
|
||||
merged_update.extend(next_update);
|
||||
}
|
||||
|
||||
assert!(
|
||||
same_state_change_source(batch_source, source_b),
|
||||
"Batch should use matching source"
|
||||
);
|
||||
assert!(same_source(batch_source, source_b.into()), "Batch should use matching source");
|
||||
assert!(merged_update.contains_key(&addr_b1));
|
||||
assert!(!merged_update.contains_key(&addr_a1));
|
||||
assert!(!merged_update.contains_key(&addr_a2));
|
||||
@@ -2255,7 +2362,7 @@ mod tests {
|
||||
|
||||
match pending_msg {
|
||||
Some(MultiProofMessage::StateUpdate(pending_source, pending_update)) => {
|
||||
assert!(same_state_change_source(pending_source, source_a));
|
||||
assert!(same_source(pending_source, source_a.into()));
|
||||
assert!(pending_update.contains_key(&addr_a2));
|
||||
}
|
||||
other => panic!("Expected pending StateUpdate with source_a, got {:?}", other),
|
||||
@@ -2298,17 +2405,20 @@ mod tests {
|
||||
|
||||
// Queue: first update dispatched immediately, next two should not merge
|
||||
let tx = task.state_root_message_sender();
|
||||
tx.send(MultiProofMessage::StateUpdate(source, create_state_update(addr1, 100))).unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(source, create_state_update(addr2, 200))).unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(source, create_state_update(addr3, 300))).unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(source.into(), create_state_update(addr1, 100)))
|
||||
.unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(source.into(), create_state_update(addr2, 200)))
|
||||
.unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(source.into(), create_state_update(addr3, 300)))
|
||||
.unwrap();
|
||||
|
||||
let mut pending_msg: Option<MultiProofMessage> = None;
|
||||
|
||||
if let Ok(MultiProofMessage::StateUpdate(first_source, first_update)) = task.rx.recv() {
|
||||
assert!(same_state_change_source(first_source, source));
|
||||
assert!(same_source(first_source, source.into()));
|
||||
assert!(first_update.contains_key(&addr1));
|
||||
|
||||
let mut accumulated_updates: Vec<(StateChangeSource, EvmState)> = Vec::new();
|
||||
let mut accumulated_updates: Vec<(Source, EvmState)> = Vec::new();
|
||||
let mut accumulated_targets = 0usize;
|
||||
|
||||
loop {
|
||||
@@ -2360,7 +2470,7 @@ mod tests {
|
||||
"Second pre-block update should not merge with a different payload"
|
||||
);
|
||||
let (batched_source, batched_update) = accumulated_updates.remove(0);
|
||||
assert!(same_state_change_source(batched_source, source));
|
||||
assert!(same_source(batched_source, source.into()));
|
||||
assert!(batched_update.contains_key(&addr2));
|
||||
assert!(!batched_update.contains_key(&addr3));
|
||||
|
||||
@@ -2440,8 +2550,8 @@ mod tests {
|
||||
let tx = task.state_root_message_sender();
|
||||
tx.send(MultiProofMessage::PrefetchProofs(targets1)).unwrap();
|
||||
tx.send(MultiProofMessage::PrefetchProofs(targets2)).unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(source, state_update1)).unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(source, state_update2)).unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(source.into(), state_update1)).unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(source.into(), state_update2)).unwrap();
|
||||
tx.send(MultiProofMessage::PrefetchProofs(targets3.clone())).unwrap();
|
||||
|
||||
// Step 1: Receive and batch PrefetchProofs (should get targets1 + targets2)
|
||||
@@ -2508,6 +2618,7 @@ mod tests {
|
||||
use revm_state::Account;
|
||||
|
||||
let test_provider_factory = create_test_provider_factory();
|
||||
let test_provider = test_provider_factory.latest().unwrap();
|
||||
let mut task = create_test_state_root_task(test_provider_factory);
|
||||
|
||||
// Queue: Prefetch1, StateUpdate, Prefetch2
|
||||
@@ -2539,7 +2650,7 @@ mod tests {
|
||||
|
||||
let tx = task.state_root_message_sender();
|
||||
tx.send(MultiProofMessage::PrefetchProofs(prefetch1)).unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(source, state_update)).unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(source.into(), state_update)).unwrap();
|
||||
tx.send(MultiProofMessage::PrefetchProofs(prefetch2.clone())).unwrap();
|
||||
|
||||
let mut ctx = MultiproofBatchCtx::new(Instant::now());
|
||||
@@ -2548,12 +2659,22 @@ mod tests {
|
||||
// First message: Prefetch1 batches; StateUpdate becomes pending.
|
||||
let first = task.rx.recv().unwrap();
|
||||
assert!(matches!(first, MultiProofMessage::PrefetchProofs(_)));
|
||||
assert!(!task.process_multiproof_message(first, &mut ctx, &mut batch_metrics));
|
||||
assert!(!task.process_multiproof_message(
|
||||
first,
|
||||
&mut ctx,
|
||||
&mut batch_metrics,
|
||||
&test_provider
|
||||
));
|
||||
let pending = ctx.pending_msg.take().expect("pending message captured");
|
||||
assert!(matches!(pending, MultiProofMessage::StateUpdate(_, _)));
|
||||
|
||||
// Pending message should be handled before the next select loop.
|
||||
assert!(!task.process_multiproof_message(pending, &mut ctx, &mut batch_metrics));
|
||||
assert!(!task.process_multiproof_message(
|
||||
pending,
|
||||
&mut ctx,
|
||||
&mut batch_metrics,
|
||||
&test_provider
|
||||
));
|
||||
|
||||
// Prefetch2 should now be in pending_msg (captured by StateUpdate's batching loop).
|
||||
match ctx.pending_msg.take() {
|
||||
@@ -2625,12 +2746,21 @@ mod tests {
|
||||
// Queue: [Prefetch1, State1, State2, State3, Prefetch2]
|
||||
let tx = task.state_root_message_sender();
|
||||
tx.send(MultiProofMessage::PrefetchProofs(prefetch1.clone())).unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(source, create_state_update(state_addr1, 100)))
|
||||
.unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(source, create_state_update(state_addr2, 200)))
|
||||
.unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(source, create_state_update(state_addr3, 300)))
|
||||
.unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(
|
||||
source.into(),
|
||||
create_state_update(state_addr1, 100),
|
||||
))
|
||||
.unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(
|
||||
source.into(),
|
||||
create_state_update(state_addr2, 200),
|
||||
))
|
||||
.unwrap();
|
||||
tx.send(MultiProofMessage::StateUpdate(
|
||||
source.into(),
|
||||
create_state_update(state_addr3, 300),
|
||||
))
|
||||
.unwrap();
|
||||
tx.send(MultiProofMessage::PrefetchProofs(prefetch2.clone())).unwrap();
|
||||
|
||||
// Simulate the state-machine loop behavior
|
||||
@@ -2703,4 +2833,44 @@ mod tests {
|
||||
_ => panic!("Prefetch2 was lost!"),
|
||||
}
|
||||
}
|
||||
|
||||
/// Verifies that BAL messages are processed correctly and generate state updates.
|
||||
#[test]
|
||||
fn test_bal_message_processing() {
|
||||
let test_provider_factory = create_test_provider_factory();
|
||||
let test_provider = test_provider_factory.latest().unwrap();
|
||||
let mut task = create_test_state_root_task(test_provider_factory);
|
||||
|
||||
// Create a simple BAL with one account change
|
||||
let account_address = Address::random();
|
||||
let account_changes = AccountChanges {
|
||||
address: account_address,
|
||||
balance_changes: vec![BalanceChange::new(0, U256::from(1000))],
|
||||
nonce_changes: vec![],
|
||||
code_changes: vec![],
|
||||
storage_changes: vec![],
|
||||
storage_reads: vec![],
|
||||
};
|
||||
|
||||
let bal = Arc::new(vec![account_changes]);
|
||||
|
||||
let mut ctx = MultiproofBatchCtx::new(Instant::now());
|
||||
let mut batch_metrics = MultiproofBatchMetrics::default();
|
||||
|
||||
let should_finish = task.process_multiproof_message(
|
||||
MultiProofMessage::BlockAccessList(bal),
|
||||
&mut ctx,
|
||||
&mut batch_metrics,
|
||||
&test_provider,
|
||||
);
|
||||
|
||||
// BAL should mark updates as finished
|
||||
assert!(ctx.updates_finished_time.is_some());
|
||||
|
||||
// Should have dispatched state update proofs
|
||||
assert!(batch_metrics.state_update_proofs_requested > 0);
|
||||
|
||||
// Should need to wait for the results of those proofs to arrive
|
||||
assert!(!should_finish, "Should continue waiting for proofs");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -402,6 +402,14 @@ where
|
||||
// use prewarming background task
|
||||
let txs = self.tx_iterator_for(&input)?;
|
||||
|
||||
// Extract the BAL, if valid and available
|
||||
let block_access_list = ensure_ok!(input
|
||||
.block_access_list()
|
||||
.transpose()
|
||||
// Eventually gets converted to a `InsertBlockErrorKind::Other`
|
||||
.map_err(Box::<dyn std::error::Error + Send + Sync>::from))
|
||||
.map(Arc::new);
|
||||
|
||||
// Spawn the appropriate processor based on strategy
|
||||
let mut handle = ensure_ok!(self.spawn_payload_processor(
|
||||
env.clone(),
|
||||
@@ -410,6 +418,7 @@ where
|
||||
parent_hash,
|
||||
ctx.state(),
|
||||
strategy,
|
||||
block_access_list,
|
||||
));
|
||||
|
||||
// Use cached state provider before executing, used in execution after prewarming threads
|
||||
@@ -779,6 +788,7 @@ where
|
||||
parent_hash: B256,
|
||||
state: &EngineApiTreeState<N>,
|
||||
strategy: StateRootStrategy,
|
||||
block_access_list: Option<Arc<BlockAccessList>>,
|
||||
) -> Result<
|
||||
PayloadHandle<
|
||||
impl ExecutableTxFor<Evm> + use<N, P, Evm, V, T>,
|
||||
@@ -808,12 +818,14 @@ where
|
||||
.record(trie_input_start.elapsed().as_secs_f64());
|
||||
|
||||
let spawn_start = Instant::now();
|
||||
|
||||
let handle = self.payload_processor.spawn(
|
||||
env,
|
||||
txs,
|
||||
provider_builder,
|
||||
multiproof_provider_factory,
|
||||
&self.config,
|
||||
block_access_list,
|
||||
);
|
||||
|
||||
// record prewarming initialization duration
|
||||
|
||||
@@ -150,6 +150,12 @@ where
|
||||
let era1_id = Era1Id::new(&config.network, start_block, block_count as u32)
|
||||
.with_hash(historical_root);
|
||||
|
||||
let era1_id = if config.max_blocks_per_file == MAX_BLOCKS_PER_ERA1 as u64 {
|
||||
era1_id
|
||||
} else {
|
||||
era1_id.with_era_count()
|
||||
};
|
||||
|
||||
debug!("Final file name {}", era1_id.to_file_name());
|
||||
let file_path = config.dir.join(era1_id.to_file_name());
|
||||
let file = std::fs::File::create(&file_path)?;
|
||||
|
||||
@@ -24,7 +24,7 @@ fn test_export_with_genesis_only() {
|
||||
assert!(file_path.exists(), "Exported file should exist on disk");
|
||||
let file_name = file_path.file_name().unwrap().to_str().unwrap();
|
||||
assert!(
|
||||
file_name.starts_with("mainnet-00000-00001-"),
|
||||
file_name.starts_with("mainnet-00000-"),
|
||||
"File should have correct prefix with era format"
|
||||
);
|
||||
assert!(file_name.ends_with(".era1"), "File should have correct extension");
|
||||
|
||||
@@ -30,8 +30,11 @@ pub trait EraFileFormat: Sized {
|
||||
|
||||
/// Era file identifiers
|
||||
pub trait EraFileId: Clone {
|
||||
/// Convert to standardized file name
|
||||
fn to_file_name(&self) -> String;
|
||||
/// File type for this identifier
|
||||
const FILE_TYPE: EraFileType;
|
||||
|
||||
/// Number of items, slots for `era`, blocks for `era1`, per era
|
||||
const ITEMS_PER_ERA: u64;
|
||||
|
||||
/// Get the network name
|
||||
fn network_name(&self) -> &str;
|
||||
@@ -41,6 +44,43 @@ pub trait EraFileId: Clone {
|
||||
|
||||
/// Get the count of items
|
||||
fn count(&self) -> u32;
|
||||
|
||||
/// Get the optional hash identifier
|
||||
fn hash(&self) -> Option<[u8; 4]>;
|
||||
|
||||
/// Whether to include era count in filename
|
||||
fn include_era_count(&self) -> bool;
|
||||
|
||||
/// Calculate era number
|
||||
fn era_number(&self) -> u64 {
|
||||
self.start_number() / Self::ITEMS_PER_ERA
|
||||
}
|
||||
|
||||
/// Calculate the number of eras spanned per file.
|
||||
///
|
||||
/// If the user can decide how many slots/blocks per era file there are, we need to calculate
|
||||
/// it. Most of the time it should be 1, but it can never be more than 2 eras per file
|
||||
/// as there is a maximum of 8192 slots/blocks per era file.
|
||||
fn era_count(&self) -> u64 {
|
||||
if self.count() == 0 {
|
||||
return 0;
|
||||
}
|
||||
let first_era = self.era_number();
|
||||
let last_number = self.start_number() + self.count() as u64 - 1;
|
||||
let last_era = last_number / Self::ITEMS_PER_ERA;
|
||||
last_era - first_era + 1
|
||||
}
|
||||
|
||||
/// Convert to standardized file name.
|
||||
fn to_file_name(&self) -> String {
|
||||
Self::FILE_TYPE.format_filename(
|
||||
self.network_name(),
|
||||
self.era_number(),
|
||||
self.hash(),
|
||||
self.include_era_count(),
|
||||
self.era_count(),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/// [`StreamReader`] for reading era-format files
|
||||
@@ -154,6 +194,37 @@ impl EraFileType {
|
||||
}
|
||||
}
|
||||
|
||||
/// Generate era file name.
|
||||
///
|
||||
/// Standard format: `<config-name>-<era-number>-<short-historical-root>.<ext>`
|
||||
/// See also <https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era.md#file-name>
|
||||
///
|
||||
/// With era count (for custom exports):
|
||||
/// `<config-name>-<era-number>-<era-count>-<short-historical-root>.<ext>`
|
||||
pub fn format_filename(
|
||||
&self,
|
||||
network_name: &str,
|
||||
era_number: u64,
|
||||
hash: Option<[u8; 4]>,
|
||||
include_era_count: bool,
|
||||
era_count: u64,
|
||||
) -> String {
|
||||
let hash = format_hash(hash);
|
||||
|
||||
if include_era_count {
|
||||
format!(
|
||||
"{}-{:05}-{:05}-{}{}",
|
||||
network_name,
|
||||
era_number,
|
||||
era_count,
|
||||
hash,
|
||||
self.extension()
|
||||
)
|
||||
} else {
|
||||
format!("{}-{:05}-{}{}", network_name, era_number, hash, self.extension())
|
||||
}
|
||||
}
|
||||
|
||||
/// Detect file type from URL
|
||||
/// By default, it assumes `Era` type
|
||||
pub fn from_url(url: &str) -> Self {
|
||||
@@ -164,3 +235,11 @@ impl EraFileType {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Format hash as hex string, or placeholder if none
|
||||
pub fn format_hash(hash: Option<[u8; 4]>) -> String {
|
||||
match hash {
|
||||
Some(h) => format!("{:02x}{:02x}{:02x}{:02x}", h[0], h[1], h[2], h[3]),
|
||||
None => "00000000".to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
//! See also <https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era.md>
|
||||
|
||||
use crate::{
|
||||
common::file_ops::EraFileId,
|
||||
common::file_ops::{EraFileId, EraFileType},
|
||||
e2s::types::{Entry, IndexEntry, SLOT_INDEX},
|
||||
era::types::consensus::{CompressedBeaconState, CompressedSignedBeaconBlock},
|
||||
};
|
||||
@@ -163,12 +163,22 @@ pub struct EraId {
|
||||
/// Optional hash identifier for this file
|
||||
/// First 4 bytes of the last historical root in the last state in the era file
|
||||
pub hash: Option<[u8; 4]>,
|
||||
|
||||
/// Whether to include era count in filename
|
||||
/// It is used for custom exports when we don't use the max number of items per file
|
||||
include_era_count: bool,
|
||||
}
|
||||
|
||||
impl EraId {
|
||||
/// Create a new [`EraId`]
|
||||
pub fn new(network_name: impl Into<String>, start_slot: u64, slot_count: u32) -> Self {
|
||||
Self { network_name: network_name.into(), start_slot, slot_count, hash: None }
|
||||
Self {
|
||||
network_name: network_name.into(),
|
||||
start_slot,
|
||||
slot_count,
|
||||
hash: None,
|
||||
include_era_count: false,
|
||||
}
|
||||
}
|
||||
|
||||
/// Add a hash identifier to [`EraId`]
|
||||
@@ -177,32 +187,18 @@ impl EraId {
|
||||
self
|
||||
}
|
||||
|
||||
/// Calculate which era number the file starts at
|
||||
pub const fn era_number(&self) -> u64 {
|
||||
self.start_slot / SLOTS_PER_HISTORICAL_ROOT
|
||||
}
|
||||
|
||||
// Helper function to calculate the number of eras per era1 file,
|
||||
// If the user can decide how many blocks per era1 file there are, we need to calculate it.
|
||||
// Most of the time it should be 1, but it can never be more than 2 eras per file
|
||||
// as there is a maximum of 8192 blocks per era1 file.
|
||||
const fn calculate_era_count(&self) -> u64 {
|
||||
if self.slot_count == 0 {
|
||||
return 0;
|
||||
}
|
||||
|
||||
let first_era = self.era_number();
|
||||
|
||||
// Calculate the actual last slot number in the range
|
||||
let last_slot = self.start_slot + self.slot_count as u64 - 1;
|
||||
// Find which era the last block belongs to
|
||||
let last_era = last_slot / SLOTS_PER_HISTORICAL_ROOT;
|
||||
// Count how many eras we span
|
||||
last_era - first_era + 1
|
||||
/// Include era count in filename, for custom slot-per-file exports
|
||||
pub const fn with_era_count(mut self) -> Self {
|
||||
self.include_era_count = true;
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
impl EraFileId for EraId {
|
||||
const FILE_TYPE: EraFileType = EraFileType::Era;
|
||||
|
||||
const ITEMS_PER_ERA: u64 = SLOTS_PER_HISTORICAL_ROOT;
|
||||
|
||||
fn network_name(&self) -> &str {
|
||||
&self.network_name
|
||||
}
|
||||
@@ -214,24 +210,13 @@ impl EraFileId for EraId {
|
||||
fn count(&self) -> u32 {
|
||||
self.slot_count
|
||||
}
|
||||
/// Convert to file name following the era file naming:
|
||||
/// `<config-name>-<era-number>-<era-count>-<short-historical-root>.era`
|
||||
/// <https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era.md#file-name>
|
||||
/// See also <https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era.md>
|
||||
fn to_file_name(&self) -> String {
|
||||
let era_number = self.era_number();
|
||||
let era_count = self.calculate_era_count();
|
||||
|
||||
if let Some(hash) = self.hash {
|
||||
format!(
|
||||
"{}-{:05}-{:05}-{:02x}{:02x}{:02x}{:02x}.era",
|
||||
self.network_name, era_number, era_count, hash[0], hash[1], hash[2], hash[3]
|
||||
)
|
||||
} else {
|
||||
// era spec format with placeholder hash when no hash available
|
||||
// Format: `<config-name>-<era-number>-<era-count>-00000000.era`
|
||||
format!("{}-{:05}-{:05}-00000000.era", self.network_name, era_number, era_count)
|
||||
}
|
||||
fn hash(&self) -> Option<[u8; 4]> {
|
||||
self.hash
|
||||
}
|
||||
|
||||
fn include_era_count(&self) -> bool {
|
||||
self.include_era_count
|
||||
}
|
||||
}
|
||||
|
||||
@@ -399,4 +384,40 @@ mod tests {
|
||||
let parsed_offset = index.offsets[0];
|
||||
assert_eq!(parsed_offset, -1024);
|
||||
}
|
||||
|
||||
#[test_case::test_case(
|
||||
EraId::new("mainnet", 0, 8192).with_hash([0x4b, 0x36, 0x3d, 0xb9]),
|
||||
"mainnet-00000-4b363db9.era";
|
||||
"Mainnet era 0"
|
||||
)]
|
||||
#[test_case::test_case(
|
||||
EraId::new("mainnet", 8192, 8192).with_hash([0x40, 0xcf, 0x2f, 0x3c]),
|
||||
"mainnet-00001-40cf2f3c.era";
|
||||
"Mainnet era 1"
|
||||
)]
|
||||
#[test_case::test_case(
|
||||
EraId::new("mainnet", 0, 8192),
|
||||
"mainnet-00000-00000000.era";
|
||||
"Without hash"
|
||||
)]
|
||||
fn test_era_id_file_naming(id: EraId, expected_file_name: &str) {
|
||||
let actual_file_name = id.to_file_name();
|
||||
assert_eq!(actual_file_name, expected_file_name);
|
||||
}
|
||||
|
||||
// File naming with era-count, for custom exports
|
||||
#[test_case::test_case(
|
||||
EraId::new("mainnet", 0, 8192).with_hash([0x4b, 0x36, 0x3d, 0xb9]).with_era_count(),
|
||||
"mainnet-00000-00001-4b363db9.era";
|
||||
"Mainnet era 0 with count"
|
||||
)]
|
||||
#[test_case::test_case(
|
||||
EraId::new("mainnet", 8000, 500).with_hash([0xab, 0xcd, 0xef, 0x12]).with_era_count(),
|
||||
"mainnet-00000-00002-abcdef12.era";
|
||||
"Spanning two eras with count"
|
||||
)]
|
||||
fn test_era_id_file_naming_with_era_count(id: EraId, expected_file_name: &str) {
|
||||
let actual_file_name = id.to_file_name();
|
||||
assert_eq!(actual_file_name, expected_file_name);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
//! See also <https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era1.md>
|
||||
|
||||
use crate::{
|
||||
common::file_ops::EraFileId,
|
||||
common::file_ops::{EraFileId, EraFileType},
|
||||
e2s::types::{Entry, IndexEntry},
|
||||
era1::types::execution::{Accumulator, BlockTuple, MAX_BLOCKS_PER_ERA1},
|
||||
};
|
||||
@@ -105,6 +105,10 @@ pub struct Era1Id {
|
||||
/// Optional hash identifier for this file
|
||||
/// First 4 bytes of the last historical root in the last state in the era file
|
||||
pub hash: Option<[u8; 4]>,
|
||||
|
||||
/// Whether to include era count in filename
|
||||
/// It is used for custom exports when we don't use the max number of items per file
|
||||
pub include_era_count: bool,
|
||||
}
|
||||
|
||||
impl Era1Id {
|
||||
@@ -114,7 +118,13 @@ impl Era1Id {
|
||||
start_block: BlockNumber,
|
||||
block_count: u32,
|
||||
) -> Self {
|
||||
Self { network_name: network_name.into(), start_block, block_count, hash: None }
|
||||
Self {
|
||||
network_name: network_name.into(),
|
||||
start_block,
|
||||
block_count,
|
||||
hash: None,
|
||||
include_era_count: false,
|
||||
}
|
||||
}
|
||||
|
||||
/// Add a hash identifier to [`Era1Id`]
|
||||
@@ -123,21 +133,17 @@ impl Era1Id {
|
||||
self
|
||||
}
|
||||
|
||||
// Helper function to calculate the number of eras per era1 file,
|
||||
// If the user can decide how many blocks per era1 file there are, we need to calculate it.
|
||||
// Most of the time it should be 1, but it can never be more than 2 eras per file
|
||||
// as there is a maximum of 8192 blocks per era1 file.
|
||||
const fn calculate_era_count(&self, first_era: u64) -> u64 {
|
||||
// Calculate the actual last block number in the range
|
||||
let last_block = self.start_block + self.block_count as u64 - 1;
|
||||
// Find which era the last block belongs to
|
||||
let last_era = last_block / MAX_BLOCKS_PER_ERA1 as u64;
|
||||
// Count how many eras we span
|
||||
last_era - first_era + 1
|
||||
/// Include era count in filename, for custom block-per-file exports
|
||||
pub const fn with_era_count(mut self) -> Self {
|
||||
self.include_era_count = true;
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
impl EraFileId for Era1Id {
|
||||
const FILE_TYPE: EraFileType = EraFileType::Era1;
|
||||
|
||||
const ITEMS_PER_ERA: u64 = MAX_BLOCKS_PER_ERA1 as u64;
|
||||
fn network_name(&self) -> &str {
|
||||
&self.network_name
|
||||
}
|
||||
@@ -149,24 +155,13 @@ impl EraFileId for Era1Id {
|
||||
fn count(&self) -> u32 {
|
||||
self.block_count
|
||||
}
|
||||
/// Convert to file name following the era file naming:
|
||||
/// `<config-name>-<era-number>-<era-count>-<short-historical-root>.era(1)`
|
||||
/// <https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era.md#file-name>
|
||||
/// See also <https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era1.md>
|
||||
fn to_file_name(&self) -> String {
|
||||
// Find which era the first block belongs to
|
||||
let era_number = self.start_block / MAX_BLOCKS_PER_ERA1 as u64;
|
||||
let era_count = self.calculate_era_count(era_number);
|
||||
if let Some(hash) = self.hash {
|
||||
format!(
|
||||
"{}-{:05}-{:05}-{:02x}{:02x}{:02x}{:02x}.era1",
|
||||
self.network_name, era_number, era_count, hash[0], hash[1], hash[2], hash[3]
|
||||
)
|
||||
} else {
|
||||
// era spec format with placeholder hash when no hash available
|
||||
// Format: `<config-name>-<era-number>-<era-count>-00000000.era1`
|
||||
format!("{}-{:05}-{:05}-00000000.era1", self.network_name, era_number, era_count)
|
||||
}
|
||||
|
||||
fn hash(&self) -> Option<[u8; 4]> {
|
||||
self.hash
|
||||
}
|
||||
|
||||
fn include_era_count(&self) -> bool {
|
||||
self.include_era_count
|
||||
}
|
||||
}
|
||||
|
||||
@@ -314,35 +309,51 @@ mod tests {
|
||||
|
||||
#[test_case::test_case(
|
||||
Era1Id::new("mainnet", 0, 8192).with_hash([0x5e, 0xc1, 0xff, 0xb8]),
|
||||
"mainnet-00000-00001-5ec1ffb8.era1";
|
||||
"mainnet-00000-5ec1ffb8.era1";
|
||||
"Mainnet era 0"
|
||||
)]
|
||||
#[test_case::test_case(
|
||||
Era1Id::new("mainnet", 8192, 8192).with_hash([0x5e, 0xcb, 0x9b, 0xf9]),
|
||||
"mainnet-00001-00001-5ecb9bf9.era1";
|
||||
"mainnet-00001-5ecb9bf9.era1";
|
||||
"Mainnet era 1"
|
||||
)]
|
||||
#[test_case::test_case(
|
||||
Era1Id::new("sepolia", 0, 8192).with_hash([0x90, 0x91, 0x84, 0x72]),
|
||||
"sepolia-00000-00001-90918472.era1";
|
||||
"sepolia-00000-90918472.era1";
|
||||
"Sepolia era 0"
|
||||
)]
|
||||
#[test_case::test_case(
|
||||
Era1Id::new("sepolia", 155648, 8192).with_hash([0xfa, 0x77, 0x00, 0x19]),
|
||||
"sepolia-00019-00001-fa770019.era1";
|
||||
"sepolia-00019-fa770019.era1";
|
||||
"Sepolia era 19"
|
||||
)]
|
||||
#[test_case::test_case(
|
||||
Era1Id::new("mainnet", 1000, 100),
|
||||
"mainnet-00000-00001-00000000.era1";
|
||||
"mainnet-00000-00000000.era1";
|
||||
"ID without hash"
|
||||
)]
|
||||
#[test_case::test_case(
|
||||
Era1Id::new("sepolia", 101130240, 8192).with_hash([0xab, 0xcd, 0xef, 0x12]),
|
||||
"sepolia-12345-00001-abcdef12.era1";
|
||||
"sepolia-12345-abcdef12.era1";
|
||||
"Large block number era 12345"
|
||||
)]
|
||||
fn test_era1id_file_naming(id: Era1Id, expected_file_name: &str) {
|
||||
fn test_era1_id_file_naming(id: Era1Id, expected_file_name: &str) {
|
||||
let actual_file_name = id.to_file_name();
|
||||
assert_eq!(actual_file_name, expected_file_name);
|
||||
}
|
||||
|
||||
// File naming with era-count, for custom exports
|
||||
#[test_case::test_case(
|
||||
Era1Id::new("mainnet", 0, 8192).with_hash([0x5e, 0xc1, 0xff, 0xb8]).with_era_count(),
|
||||
"mainnet-00000-00001-5ec1ffb8.era1";
|
||||
"Mainnet era 0 with count"
|
||||
)]
|
||||
#[test_case::test_case(
|
||||
Era1Id::new("mainnet", 8000, 500).with_hash([0xab, 0xcd, 0xef, 0x12]).with_era_count(),
|
||||
"mainnet-00000-00002-abcdef12.era1";
|
||||
"Spanning two eras with count"
|
||||
)]
|
||||
fn test_era1_id_file_naming_with_era_count(id: Era1Id, expected_file_name: &str) {
|
||||
let actual_file_name = id.to_file_name();
|
||||
assert_eq!(actual_file_name, expected_file_name);
|
||||
}
|
||||
|
||||
@@ -88,6 +88,9 @@ asm-keccak = [
|
||||
"reth-node-core/asm-keccak",
|
||||
"revm/asm-keccak",
|
||||
]
|
||||
keccak-cache-global = [
|
||||
"alloy-primitives/keccak-cache-global",
|
||||
]
|
||||
js-tracer = [
|
||||
"reth-node-builder/js-tracer",
|
||||
"reth-rpc/js-tracer",
|
||||
|
||||
@@ -79,7 +79,9 @@ arbitrary = [
|
||||
"alloy-rpc-types-engine?/arbitrary",
|
||||
"reth-codecs?/arbitrary",
|
||||
]
|
||||
|
||||
keccak-cache-global = [
|
||||
"reth-node-ethereum?/keccak-cache-global",
|
||||
]
|
||||
test-utils = [
|
||||
"reth-chainspec/test-utils",
|
||||
"reth-consensus?/test-utils",
|
||||
|
||||
@@ -1218,7 +1218,9 @@ impl ReverseHeadersDownloaderBuilder {
|
||||
next_request_block_number: 0,
|
||||
next_chain_tip_block_number: 0,
|
||||
lowest_validated_header: None,
|
||||
request_limit,
|
||||
// TODO(mattsse): tmp hotfix to prevent issues with syncing from besu which has an upper
|
||||
// limit of 512
|
||||
request_limit: request_limit.min(512),
|
||||
min_concurrent_requests,
|
||||
max_concurrent_requests,
|
||||
stream_batch_size,
|
||||
|
||||
@@ -373,13 +373,13 @@ impl<N: NetworkPrimitives> EthMessage<N> {
|
||||
///
|
||||
/// This handles up/downcasting where appropriate, for example for different receipt request
|
||||
/// types.
|
||||
pub fn map_versioned(mut self, version: EthVersion) -> Self {
|
||||
pub fn map_versioned(self, version: EthVersion) -> Self {
|
||||
// For eth/70 peers we send `GetReceipts` using the new eth/70
|
||||
// encoding with `firstBlockReceiptIndex = 0`, while keeping the
|
||||
// user-facing `PeerRequest` API unchanged.
|
||||
if version >= EthVersion::Eth70 {
|
||||
return match self {
|
||||
EthMessage::GetReceipts(pair) => {
|
||||
Self::GetReceipts(pair) => {
|
||||
let RequestPair { request_id, message } = pair;
|
||||
let req = RequestPair {
|
||||
request_id,
|
||||
@@ -388,7 +388,7 @@ impl<N: NetworkPrimitives> EthMessage<N> {
|
||||
block_hashes: message.0,
|
||||
},
|
||||
};
|
||||
EthMessage::GetReceipts70(req)
|
||||
Self::GetReceipts70(req)
|
||||
}
|
||||
other => other,
|
||||
}
|
||||
|
||||
@@ -218,6 +218,9 @@ where
|
||||
let _ = response.send(Ok(Receipts69(receipts)));
|
||||
}
|
||||
|
||||
/// Handles partial responses for [`GetReceipts70`] queries.
|
||||
///
|
||||
/// This will adhere to the soft limit but allow filling the last vec partially.
|
||||
fn on_receipts70_request(
|
||||
&self,
|
||||
_peer_id: PeerId,
|
||||
|
||||
@@ -318,7 +318,7 @@ impl<N: NetworkPrimitives> ActiveSession<N> {
|
||||
fn on_internal_peer_request(&mut self, request: PeerRequest<N>, deadline: Instant) {
|
||||
let request_id = self.next_id();
|
||||
trace!(?request, peer_id=?self.remote_peer_id, ?request_id, "sending request to peer");
|
||||
let mut msg = request.create_request_message(request_id).map_versioned(self.conn.version());
|
||||
let msg = request.create_request_message(request_id).map_versioned(self.conn.version());
|
||||
|
||||
self.queued_outgoing.push_back(msg.into());
|
||||
let req = InflightRequest {
|
||||
@@ -374,7 +374,6 @@ impl<N: NetworkPrimitives> ActiveSession<N> {
|
||||
fn handle_outgoing_response(&mut self, id: u64, resp: PeerResponseResult<N>) {
|
||||
match resp.try_into_message(id) {
|
||||
Ok(msg) => {
|
||||
let msg: EthMessage<N> = msg;
|
||||
self.queued_outgoing.push_back(msg.into());
|
||||
}
|
||||
Err(err) => {
|
||||
|
||||
@@ -485,8 +485,9 @@ where
|
||||
.with_blocks_per_file_for_segments(static_files_config.as_blocks_per_file_map())
|
||||
.build()?;
|
||||
|
||||
// Initialize RocksDB provider with metrics and statistics enabled
|
||||
// Initialize RocksDB provider with metrics, statistics, and default tables
|
||||
let rocksdb_provider = RocksDBProvider::builder(self.data_dir().rocksdb())
|
||||
.with_default_tables()
|
||||
.with_metrics()
|
||||
.with_statistics()
|
||||
.build()?;
|
||||
|
||||
@@ -27,7 +27,7 @@ tracing.workspace = true
|
||||
workspace = true
|
||||
|
||||
[features]
|
||||
default = ["jemalloc", "otlp", "reth-optimism-evm/portable", "js-tracer"]
|
||||
default = ["jemalloc", "otlp", "reth-optimism-evm/portable", "js-tracer", "keccak-cache-global"]
|
||||
|
||||
otlp = ["reth-optimism-cli/otlp"]
|
||||
|
||||
@@ -40,7 +40,9 @@ jemalloc-prof = ["reth-cli-util/jemalloc-prof"]
|
||||
tracy-allocator = ["reth-cli-util/tracy-allocator"]
|
||||
|
||||
asm-keccak = ["reth-optimism-cli/asm-keccak", "reth-optimism-node/asm-keccak"]
|
||||
|
||||
keccak-cache-global = [
|
||||
"reth-optimism-node/keccak-cache-global",
|
||||
]
|
||||
dev = [
|
||||
"reth-optimism-cli/dev",
|
||||
"reth-optimism-primitives/arbitrary",
|
||||
|
||||
@@ -93,6 +93,10 @@ asm-keccak = [
|
||||
"reth-node-core/asm-keccak",
|
||||
"revm/asm-keccak",
|
||||
]
|
||||
keccak-cache-global = [
|
||||
"alloy-primitives/keccak-cache-global",
|
||||
"reth-optimism-node/keccak-cache-global",
|
||||
]
|
||||
js-tracer = [
|
||||
"reth-node-builder/js-tracer",
|
||||
"reth-optimism-node/js-tracer",
|
||||
|
||||
@@ -74,7 +74,9 @@ arbitrary = [
|
||||
"reth-eth-wire?/arbitrary",
|
||||
"reth-codecs?/arbitrary",
|
||||
]
|
||||
|
||||
keccak-cache-global = [
|
||||
"reth-optimism-node?/keccak-cache-global",
|
||||
]
|
||||
test-utils = [
|
||||
"reth-chainspec/test-utils",
|
||||
"reth-consensus?/test-utils",
|
||||
|
||||
@@ -218,7 +218,7 @@ impl<B: Block> RecoveredBlock<B> {
|
||||
|
||||
/// A safer variant of [`Self::new_sealed`] that checks if the number of senders is equal to
|
||||
/// the number of transactions in the block and recovers the senders from the transactions, if
|
||||
/// not using [`SignedTransaction::recover_signer_unchecked`](crate::transaction::signed::SignedTransaction)
|
||||
/// not using [`SignedTransaction::recover_signer`](crate::transaction::signed::SignedTransaction)
|
||||
/// to recover the senders.
|
||||
///
|
||||
/// Returns an error if any of the transactions fail to recover the sender.
|
||||
|
||||
@@ -175,8 +175,6 @@ where
|
||||
// need to apply the state changes of this call before executing the
|
||||
// next call
|
||||
if calls.peek().is_some() {
|
||||
// need to apply the state changes of this call before executing
|
||||
// the next call
|
||||
db.commit(res.state)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -9,14 +9,19 @@ use crate::{
|
||||
providers::{StaticFileProvider, StaticFileProviderRWRefMut},
|
||||
StaticFileProviderFactory,
|
||||
};
|
||||
use alloy_primitives::{map::HashMap, Address, BlockNumber, TxNumber};
|
||||
use alloy_primitives::{map::HashMap, Address, BlockNumber, TxHash, TxNumber};
|
||||
use reth_db::{
|
||||
cursor::DbCursorRO,
|
||||
static_file::TransactionSenderMask,
|
||||
table::Value,
|
||||
transaction::{CursorMutTy, CursorTy, DbTx, DbTxMut},
|
||||
};
|
||||
use reth_db_api::{cursor::DbCursorRW, tables};
|
||||
use reth_db_api::{
|
||||
cursor::DbCursorRW,
|
||||
models::{storage_sharded_key::StorageShardedKey, ShardedKey},
|
||||
tables,
|
||||
tables::BlockNumberList,
|
||||
};
|
||||
use reth_errors::ProviderError;
|
||||
use reth_node_types::NodePrimitives;
|
||||
use reth_primitives_traits::ReceiptTy;
|
||||
@@ -294,6 +299,108 @@ where
|
||||
}
|
||||
}
|
||||
|
||||
impl<'a, CURSOR, N: NodePrimitives> EitherWriter<'a, CURSOR, N>
|
||||
where
|
||||
CURSOR: DbCursorRW<tables::TransactionHashNumbers> + DbCursorRO<tables::TransactionHashNumbers>,
|
||||
{
|
||||
/// Puts a transaction hash number mapping.
|
||||
pub fn put_transaction_hash_number(
|
||||
&mut self,
|
||||
hash: TxHash,
|
||||
tx_num: TxNumber,
|
||||
) -> ProviderResult<()> {
|
||||
match self {
|
||||
Self::Database(cursor) => Ok(cursor.upsert(hash, &tx_num)?),
|
||||
Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
|
||||
#[cfg(all(unix, feature = "rocksdb"))]
|
||||
Self::RocksDB(batch) => batch.put::<tables::TransactionHashNumbers>(hash, &tx_num),
|
||||
}
|
||||
}
|
||||
|
||||
/// Deletes a transaction hash number mapping.
|
||||
pub fn delete_transaction_hash_number(&mut self, hash: TxHash) -> ProviderResult<()> {
|
||||
match self {
|
||||
Self::Database(cursor) => {
|
||||
if cursor.seek_exact(hash)?.is_some() {
|
||||
cursor.delete_current()?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
|
||||
#[cfg(all(unix, feature = "rocksdb"))]
|
||||
Self::RocksDB(batch) => batch.delete::<tables::TransactionHashNumbers>(hash),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<'a, CURSOR, N: NodePrimitives> EitherWriter<'a, CURSOR, N>
|
||||
where
|
||||
CURSOR: DbCursorRW<tables::StoragesHistory> + DbCursorRO<tables::StoragesHistory>,
|
||||
{
|
||||
/// Puts a storage history entry.
|
||||
pub fn put_storage_history(
|
||||
&mut self,
|
||||
key: StorageShardedKey,
|
||||
value: &BlockNumberList,
|
||||
) -> ProviderResult<()> {
|
||||
match self {
|
||||
Self::Database(cursor) => Ok(cursor.upsert(key, value)?),
|
||||
Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
|
||||
#[cfg(all(unix, feature = "rocksdb"))]
|
||||
Self::RocksDB(batch) => batch.put::<tables::StoragesHistory>(key, value),
|
||||
}
|
||||
}
|
||||
|
||||
/// Deletes a storage history entry.
|
||||
pub fn delete_storage_history(&mut self, key: StorageShardedKey) -> ProviderResult<()> {
|
||||
match self {
|
||||
Self::Database(cursor) => {
|
||||
if cursor.seek_exact(key)?.is_some() {
|
||||
cursor.delete_current()?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
|
||||
#[cfg(all(unix, feature = "rocksdb"))]
|
||||
Self::RocksDB(batch) => batch.delete::<tables::StoragesHistory>(key),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<'a, CURSOR, N: NodePrimitives> EitherWriter<'a, CURSOR, N>
|
||||
where
|
||||
CURSOR: DbCursorRW<tables::AccountsHistory> + DbCursorRO<tables::AccountsHistory>,
|
||||
{
|
||||
/// Puts an account history entry.
|
||||
pub fn put_account_history(
|
||||
&mut self,
|
||||
key: ShardedKey<Address>,
|
||||
value: &BlockNumberList,
|
||||
) -> ProviderResult<()> {
|
||||
match self {
|
||||
Self::Database(cursor) => Ok(cursor.upsert(key, value)?),
|
||||
Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
|
||||
#[cfg(all(unix, feature = "rocksdb"))]
|
||||
Self::RocksDB(batch) => batch.put::<tables::AccountsHistory>(key, value),
|
||||
}
|
||||
}
|
||||
|
||||
/// Deletes an account history entry.
|
||||
pub fn delete_account_history(&mut self, key: ShardedKey<Address>) -> ProviderResult<()> {
|
||||
match self {
|
||||
Self::Database(cursor) => {
|
||||
if cursor.seek_exact(key)?.is_some() {
|
||||
cursor.delete_current()?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
|
||||
#[cfg(all(unix, feature = "rocksdb"))]
|
||||
Self::RocksDB(batch) => batch.delete::<tables::AccountsHistory>(key),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Represents a source for reading data, either from database, static files, or `RocksDB`.
|
||||
#[derive(Debug, Display)]
|
||||
pub enum EitherReader<'a, CURSOR, N> {
|
||||
@@ -418,6 +525,60 @@ where
|
||||
}
|
||||
}
|
||||
|
||||
impl<CURSOR, N: NodePrimitives> EitherReader<'_, CURSOR, N>
|
||||
where
|
||||
CURSOR: DbCursorRO<tables::TransactionHashNumbers>,
|
||||
{
|
||||
/// Gets a transaction number by its hash.
|
||||
pub fn get_transaction_hash_number(
|
||||
&mut self,
|
||||
hash: TxHash,
|
||||
) -> ProviderResult<Option<TxNumber>> {
|
||||
match self {
|
||||
Self::Database(cursor, _) => Ok(cursor.seek_exact(hash)?.map(|(_, v)| v)),
|
||||
Self::StaticFile(_, _) => Err(ProviderError::UnsupportedProvider),
|
||||
#[cfg(all(unix, feature = "rocksdb"))]
|
||||
Self::RocksDB(tx) => tx.get::<tables::TransactionHashNumbers>(hash),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<CURSOR, N: NodePrimitives> EitherReader<'_, CURSOR, N>
|
||||
where
|
||||
CURSOR: DbCursorRO<tables::StoragesHistory>,
|
||||
{
|
||||
/// Gets a storage history entry.
|
||||
pub fn get_storage_history(
|
||||
&mut self,
|
||||
key: StorageShardedKey,
|
||||
) -> ProviderResult<Option<BlockNumberList>> {
|
||||
match self {
|
||||
Self::Database(cursor, _) => Ok(cursor.seek_exact(key)?.map(|(_, v)| v)),
|
||||
Self::StaticFile(_, _) => Err(ProviderError::UnsupportedProvider),
|
||||
#[cfg(all(unix, feature = "rocksdb"))]
|
||||
Self::RocksDB(tx) => tx.get::<tables::StoragesHistory>(key),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<CURSOR, N: NodePrimitives> EitherReader<'_, CURSOR, N>
|
||||
where
|
||||
CURSOR: DbCursorRO<tables::AccountsHistory>,
|
||||
{
|
||||
/// Gets an account history entry.
|
||||
pub fn get_account_history(
|
||||
&mut self,
|
||||
key: ShardedKey<Address>,
|
||||
) -> ProviderResult<Option<BlockNumberList>> {
|
||||
match self {
|
||||
Self::Database(cursor, _) => Ok(cursor.seek_exact(key)?.map(|(_, v)| v)),
|
||||
Self::StaticFile(_, _) => Err(ProviderError::UnsupportedProvider),
|
||||
#[cfg(all(unix, feature = "rocksdb"))]
|
||||
Self::RocksDB(tx) => tx.get::<tables::AccountsHistory>(key),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Destination for writing data.
|
||||
#[derive(Debug, EnumIs)]
|
||||
pub enum EitherWriterDestination {
|
||||
@@ -499,3 +660,160 @@ mod tests {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(all(test, unix, feature = "rocksdb"))]
|
||||
mod rocksdb_tests {
|
||||
use crate::providers::rocksdb::{RocksDBBuilder, RocksDBProvider};
|
||||
use alloy_primitives::{Address, B256};
|
||||
use reth_db_api::{
|
||||
models::{storage_sharded_key::StorageShardedKey, IntegerList, ShardedKey},
|
||||
tables,
|
||||
};
|
||||
use tempfile::TempDir;
|
||||
|
||||
fn create_rocksdb_provider() -> (TempDir, RocksDBProvider) {
|
||||
let temp_dir = TempDir::new().unwrap();
|
||||
let provider = RocksDBBuilder::new(temp_dir.path())
|
||||
.with_table::<tables::TransactionHashNumbers>()
|
||||
.with_table::<tables::StoragesHistory>()
|
||||
.with_table::<tables::AccountsHistory>()
|
||||
.build()
|
||||
.unwrap();
|
||||
(temp_dir, provider)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_rocksdb_batch_transaction_hash_numbers() {
|
||||
let (_temp_dir, provider) = create_rocksdb_provider();
|
||||
|
||||
let hash1 = B256::from([1u8; 32]);
|
||||
let hash2 = B256::from([2u8; 32]);
|
||||
let tx_num1 = 100u64;
|
||||
let tx_num2 = 200u64;
|
||||
|
||||
// Write via RocksDBBatch (same as EitherWriter::RocksDB would use internally)
|
||||
let mut batch = provider.batch();
|
||||
batch.put::<tables::TransactionHashNumbers>(hash1, &tx_num1).unwrap();
|
||||
batch.put::<tables::TransactionHashNumbers>(hash2, &tx_num2).unwrap();
|
||||
batch.commit().unwrap();
|
||||
|
||||
// Read via RocksTx (same as EitherReader::RocksDB would use internally)
|
||||
let tx = provider.tx();
|
||||
assert_eq!(tx.get::<tables::TransactionHashNumbers>(hash1).unwrap(), Some(tx_num1));
|
||||
assert_eq!(tx.get::<tables::TransactionHashNumbers>(hash2).unwrap(), Some(tx_num2));
|
||||
|
||||
// Test missing key
|
||||
let missing_hash = B256::from([99u8; 32]);
|
||||
assert_eq!(tx.get::<tables::TransactionHashNumbers>(missing_hash).unwrap(), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_rocksdb_batch_storage_history() {
|
||||
let (_temp_dir, provider) = create_rocksdb_provider();
|
||||
|
||||
let address = Address::random();
|
||||
let storage_key = B256::from([1u8; 32]);
|
||||
let key = StorageShardedKey::new(address, storage_key, 1000);
|
||||
let value = IntegerList::new([1, 5, 10, 50]).unwrap();
|
||||
|
||||
// Write via RocksDBBatch
|
||||
let mut batch = provider.batch();
|
||||
batch.put::<tables::StoragesHistory>(key.clone(), &value).unwrap();
|
||||
batch.commit().unwrap();
|
||||
|
||||
// Read via RocksTx
|
||||
let tx = provider.tx();
|
||||
let result = tx.get::<tables::StoragesHistory>(key).unwrap();
|
||||
assert_eq!(result, Some(value));
|
||||
|
||||
// Test missing key
|
||||
let missing_key = StorageShardedKey::new(Address::random(), B256::ZERO, 0);
|
||||
assert_eq!(tx.get::<tables::StoragesHistory>(missing_key).unwrap(), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_rocksdb_batch_account_history() {
|
||||
let (_temp_dir, provider) = create_rocksdb_provider();
|
||||
|
||||
let address = Address::random();
|
||||
let key = ShardedKey::new(address, 1000);
|
||||
let value = IntegerList::new([1, 10, 100, 500]).unwrap();
|
||||
|
||||
// Write via RocksDBBatch
|
||||
let mut batch = provider.batch();
|
||||
batch.put::<tables::AccountsHistory>(key.clone(), &value).unwrap();
|
||||
batch.commit().unwrap();
|
||||
|
||||
// Read via RocksTx
|
||||
let tx = provider.tx();
|
||||
let result = tx.get::<tables::AccountsHistory>(key).unwrap();
|
||||
assert_eq!(result, Some(value));
|
||||
|
||||
// Test missing key
|
||||
let missing_key = ShardedKey::new(Address::random(), 0);
|
||||
assert_eq!(tx.get::<tables::AccountsHistory>(missing_key).unwrap(), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_rocksdb_batch_delete_transaction_hash_number() {
|
||||
let (_temp_dir, provider) = create_rocksdb_provider();
|
||||
|
||||
let hash = B256::from([1u8; 32]);
|
||||
let tx_num = 100u64;
|
||||
|
||||
// First write
|
||||
provider.put::<tables::TransactionHashNumbers>(hash, &tx_num).unwrap();
|
||||
assert_eq!(provider.get::<tables::TransactionHashNumbers>(hash).unwrap(), Some(tx_num));
|
||||
|
||||
// Delete via RocksDBBatch
|
||||
let mut batch = provider.batch();
|
||||
batch.delete::<tables::TransactionHashNumbers>(hash).unwrap();
|
||||
batch.commit().unwrap();
|
||||
|
||||
// Verify deletion
|
||||
assert_eq!(provider.get::<tables::TransactionHashNumbers>(hash).unwrap(), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_rocksdb_batch_delete_storage_history() {
|
||||
let (_temp_dir, provider) = create_rocksdb_provider();
|
||||
|
||||
let address = Address::random();
|
||||
let storage_key = B256::from([1u8; 32]);
|
||||
let key = StorageShardedKey::new(address, storage_key, 1000);
|
||||
let value = IntegerList::new([1, 5, 10]).unwrap();
|
||||
|
||||
// First write
|
||||
provider.put::<tables::StoragesHistory>(key.clone(), &value).unwrap();
|
||||
assert!(provider.get::<tables::StoragesHistory>(key.clone()).unwrap().is_some());
|
||||
|
||||
// Delete via RocksDBBatch
|
||||
let mut batch = provider.batch();
|
||||
batch.delete::<tables::StoragesHistory>(key.clone()).unwrap();
|
||||
batch.commit().unwrap();
|
||||
|
||||
// Verify deletion
|
||||
assert_eq!(provider.get::<tables::StoragesHistory>(key).unwrap(), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_rocksdb_batch_delete_account_history() {
|
||||
let (_temp_dir, provider) = create_rocksdb_provider();
|
||||
|
||||
let address = Address::random();
|
||||
let key = ShardedKey::new(address, 1000);
|
||||
let value = IntegerList::new([1, 10, 100]).unwrap();
|
||||
|
||||
// First write
|
||||
provider.put::<tables::AccountsHistory>(key.clone(), &value).unwrap();
|
||||
assert!(provider.get::<tables::AccountsHistory>(key.clone()).unwrap().is_some());
|
||||
|
||||
// Delete via RocksDBBatch
|
||||
let mut batch = provider.batch();
|
||||
batch.delete::<tables::AccountsHistory>(key.clone()).unwrap();
|
||||
batch.commit().unwrap();
|
||||
|
||||
// Verify deletion
|
||||
assert_eq!(provider.get::<tables::AccountsHistory>(key).unwrap(), None);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -109,7 +109,7 @@ impl<N> ProviderFactoryBuilder<N> {
|
||||
self.db(Arc::new(open_db_read_only(db_dir, db_args)?))
|
||||
.chainspec(chainspec)
|
||||
.static_file(StaticFileProvider::read_only(static_files_dir, watch_static_files)?)
|
||||
.rocksdb_provider(RocksDBProvider::builder(&rocksdb_dir).build()?)
|
||||
.rocksdb_provider(RocksDBProvider::builder(&rocksdb_dir).with_default_tables().build()?)
|
||||
.build_provider_factory()
|
||||
.map_err(Into::into)
|
||||
}
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
use super::metrics::{RocksDBMetrics, RocksDBOperation};
|
||||
use reth_db_api::{
|
||||
table::{Compress, Decompress, Encode, Table},
|
||||
DatabaseError,
|
||||
tables, DatabaseError,
|
||||
};
|
||||
use reth_storage_errors::{
|
||||
db::{DatabaseErrorInfo, DatabaseWriteError, DatabaseWriteOperation, LogLevel},
|
||||
@@ -143,6 +143,18 @@ impl RocksDBBuilder {
|
||||
self
|
||||
}
|
||||
|
||||
/// Registers the default tables used by reth for `RocksDB` storage.
|
||||
///
|
||||
/// This registers:
|
||||
/// - [`tables::TransactionHashNumbers`] - Transaction hash to number mapping
|
||||
/// - [`tables::AccountsHistory`] - Account history index
|
||||
/// - [`tables::StoragesHistory`] - Storage history index
|
||||
pub fn with_default_tables(self) -> Self {
|
||||
self.with_table::<tables::TransactionHashNumbers>()
|
||||
.with_table::<tables::AccountsHistory>()
|
||||
.with_table::<tables::StoragesHistory>()
|
||||
}
|
||||
|
||||
/// Enables metrics.
|
||||
pub const fn with_metrics(mut self) -> Self {
|
||||
self.enable_metrics = true;
|
||||
@@ -630,10 +642,38 @@ const fn convert_log_level(level: LogLevel) -> rocksdb::LogLevel {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use alloy_primitives::{TxHash, B256};
|
||||
use reth_db_api::{table::Table, tables};
|
||||
use alloy_primitives::{Address, TxHash, B256};
|
||||
use reth_db_api::{
|
||||
models::{sharded_key::ShardedKey, storage_sharded_key::StorageShardedKey, IntegerList},
|
||||
table::Table,
|
||||
tables,
|
||||
};
|
||||
use tempfile::TempDir;
|
||||
|
||||
#[test]
|
||||
fn test_with_default_tables_registers_required_column_families() {
|
||||
let temp_dir = TempDir::new().unwrap();
|
||||
|
||||
// Build with default tables
|
||||
let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
|
||||
|
||||
// Should be able to write/read TransactionHashNumbers
|
||||
let tx_hash = TxHash::from(B256::from([1u8; 32]));
|
||||
provider.put::<tables::TransactionHashNumbers>(tx_hash, &100).unwrap();
|
||||
assert_eq!(provider.get::<tables::TransactionHashNumbers>(tx_hash).unwrap(), Some(100));
|
||||
|
||||
// Should be able to write/read AccountsHistory
|
||||
let key = ShardedKey::new(Address::ZERO, 100);
|
||||
let value = IntegerList::default();
|
||||
provider.put::<tables::AccountsHistory>(key.clone(), &value).unwrap();
|
||||
assert!(provider.get::<tables::AccountsHistory>(key).unwrap().is_some());
|
||||
|
||||
// Should be able to write/read StoragesHistory
|
||||
let key = StorageShardedKey::new(Address::ZERO, B256::ZERO, 100);
|
||||
provider.put::<tables::StoragesHistory>(key.clone(), &value).unwrap();
|
||||
assert!(provider.get::<tables::StoragesHistory>(key).unwrap().is_some());
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct TestTable;
|
||||
|
||||
|
||||
@@ -119,6 +119,11 @@ impl RocksDBBuilder {
|
||||
self
|
||||
}
|
||||
|
||||
/// Registers the default tables used by reth for `RocksDB` storage (stub implementation).
|
||||
pub const fn with_default_tables(self) -> Self {
|
||||
self
|
||||
}
|
||||
|
||||
/// Enables metrics (stub implementation).
|
||||
pub const fn with_metrics(self) -> Self {
|
||||
self
|
||||
|
||||
@@ -78,7 +78,7 @@ Logging:
|
||||
|
||||
Possible values:
|
||||
- always: Colors on
|
||||
- auto: Colors on
|
||||
- auto: Auto-detect
|
||||
- never: Colors off
|
||||
|
||||
Display:
|
||||
@@ -93,4 +93,4 @@ Display:
|
||||
|
||||
-q, --quiet
|
||||
Silence all log output
|
||||
```
|
||||
```
|
||||
|
||||
@@ -78,7 +78,7 @@ Logging:
|
||||
|
||||
Possible values:
|
||||
- always: Colors on
|
||||
- auto: Colors on
|
||||
- auto: Auto-detect
|
||||
- never: Colors off
|
||||
|
||||
Display:
|
||||
@@ -93,4 +93,4 @@ Display:
|
||||
|
||||
-q, --quiet
|
||||
Silence all log output
|
||||
```
|
||||
```
|
||||
|
||||
@@ -80,7 +80,7 @@ Logging:
|
||||
|
||||
Possible values:
|
||||
- always: Colors on
|
||||
- auto: Colors on
|
||||
- auto: Auto-detect
|
||||
- never: Colors off
|
||||
|
||||
[default: always]
|
||||
@@ -97,4 +97,4 @@ Display:
|
||||
|
||||
-q, --quiet
|
||||
Silence all log output
|
||||
```
|
||||
```
|
||||
|
||||
@@ -134,7 +134,7 @@ Logging:
|
||||
|
||||
Possible values:
|
||||
- always: Colors on
|
||||
- auto: Colors on
|
||||
- auto: Auto-detect
|
||||
- never: Colors off
|
||||
|
||||
[default: always]
|
||||
@@ -151,4 +151,4 @@ Display:
|
||||
|
||||
-q, --quiet
|
||||
Silence all log output
|
||||
```
|
||||
```
|
||||
|
||||
@@ -95,7 +95,7 @@ Logging:
|
||||
|
||||
Possible values:
|
||||
- always: Colors on
|
||||
- auto: Colors on
|
||||
- auto: Auto-detect
|
||||
- never: Colors off
|
||||
|
||||
Display:
|
||||
@@ -110,4 +110,4 @@ Display:
|
||||
|
||||
-q, --quiet
|
||||
Silence all log output
|
||||
```
|
||||
```
|
||||
|
||||
@@ -60,6 +60,15 @@ op-reth supports additional OP Stack specific CLI arguments:
|
||||
1. `--rollup.sequencer <uri>` - The sequencer endpoint to connect to. Transactions sent to the `op-reth` EL are also forwarded to this sequencer endpoint for inclusion, as the sequencer is the entity that builds blocks on OP Stack chains. Aliases: `--rollup.sequencer-http`, `--rollup.sequencer-ws`.
|
||||
1. `--rollup.disable-tx-pool-gossip` - Disables gossiping of transactions in the mempool to peers. This can be omitted for personal nodes, though providers should always opt to enable this flag.
|
||||
1. `--rollup.discovery.v4` - Enables the discovery v4 protocol for peer discovery. By default, op-reth, similar to op-geth, has discovery v5 enabled and discovery v4 disabled, whereas regular reth has discovery v4 enabled and discovery v5 disabled.
|
||||
1. `--rollup.compute-pending-block` - Enables computing of the pending block from the tx-pool instead of using the latest block. By default the pending block equals the latest block to save resources and not leak txs from the tx-pool.
|
||||
1. `--rollup.enable-tx-conditional` - Enable transaction conditional support on sequencer.
|
||||
1. `--rollup.supervisor-http <url>` - HTTP endpoint for the interop supervisor.
|
||||
1. `--rollup.supervisor-safety-level <level>` - Safety level for the supervisor (default: `CrossUnsafe`).
|
||||
1. `--rollup.sequencer-headers <headers>` - Optional headers to use when connecting to the sequencer. Requires `--rollup.sequencer`.
|
||||
1. `--rollup.historicalrpc <url>` - RPC endpoint for historical data. Alias: `--rollup.historical-rpc`.
|
||||
1. `--min-suggested-priority-fee <wei>` - Minimum suggested priority fee (tip) in wei (default: `1000000`).
|
||||
1. `--flashblocks-url <url>` - A URL pointing to a secure websocket subscription that streams out flashblocks. If given, the flashblocks are received to build pending block.
|
||||
1. `--flashblock-consensus` - Enable flashblock consensus client to drive the chain forward. Requires `--flashblocks-url`.
|
||||
|
||||
First, ensure that your L1 archival node is running and synced to tip. Also make sure that the beacon node / consensus layer client is running and has http APIs enabled. Then, start `op-reth` with the `--rollup.sequencer` flag set to the `Base Mainnet` sequencer endpoint:
|
||||
|
||||
|
||||
@@ -53,44 +53,56 @@ Provides external API access to the node:
|
||||
|
||||
## Component Customization
|
||||
|
||||
Each component can be customized through Reth's builder pattern:
|
||||
Each component can be customized through Reth's builder pattern. You can replace individual components
|
||||
while keeping the defaults for others using `EthereumNode::components()`:
|
||||
|
||||
```rust
|
||||
use reth_ethereum::node::{EthereumNode, NodeBuilder};
|
||||
use reth_ethereum::{
|
||||
cli::interface::Cli,
|
||||
node::{
|
||||
node::EthereumAddOns,
|
||||
EthereumNode,
|
||||
},
|
||||
};
|
||||
|
||||
let node = NodeBuilder::new(config)
|
||||
.with_types::<EthereumNode>()
|
||||
.with_components(|ctx| {
|
||||
// Use the ComponentBuilder to customize components
|
||||
ctx.components_builder()
|
||||
// Custom network configuration
|
||||
.network(|network_builder| {
|
||||
network_builder
|
||||
.peer_manager(custom_peer_manager)
|
||||
.build()
|
||||
})
|
||||
// Custom transaction pool
|
||||
.pool(|pool_builder| {
|
||||
pool_builder
|
||||
.validator(custom_validator)
|
||||
.ordering(custom_ordering)
|
||||
.build()
|
||||
})
|
||||
// Custom consensus
|
||||
.consensus(custom_consensus)
|
||||
// Custom EVM configuration
|
||||
.evm(|evm_builder| {
|
||||
evm_builder
|
||||
.with_precompiles(custom_precompiles)
|
||||
.build()
|
||||
})
|
||||
// Build all components
|
||||
.build()
|
||||
})
|
||||
.build()
|
||||
.await?;
|
||||
fn main() {
|
||||
Cli::parse_args()
|
||||
.run(|builder, _| async move {
|
||||
let handle = builder
|
||||
// Use the default ethereum node types
|
||||
.with_types::<EthereumNode>()
|
||||
// Configure the components of the node
|
||||
// Use default ethereum components but replace specific ones
|
||||
.with_components(
|
||||
EthereumNode::components()
|
||||
// Custom transaction pool
|
||||
.pool(CustomPoolBuilder::default())
|
||||
// Other customizable components:
|
||||
// .network(CustomNetworkBuilder::default())
|
||||
// .executor(CustomExecutorBuilder::default())
|
||||
// .consensus(CustomConsensusBuilder::default())
|
||||
// .payload(CustomPayloadBuilder::default())
|
||||
)
|
||||
.with_add_ons(EthereumAddOns::default())
|
||||
.launch()
|
||||
.await?;
|
||||
|
||||
handle.wait_for_node_exit().await
|
||||
})
|
||||
.unwrap();
|
||||
}
|
||||
```
|
||||
|
||||
Custom component builders must implement their respective traits (`PoolBuilder`, `NetworkBuilder`,
|
||||
`ExecutorBuilder`, `ConsensusBuilder`, `PayloadServiceBuilder`). Each trait requires implementing
|
||||
an async `build_*` method that receives a `BuilderContext` with access to node configuration,
|
||||
providers, and task executors.
|
||||
|
||||
For complete working examples with full trait implementations, see:
|
||||
- [custom-node-components](https://github.com/paradigmxyz/reth/tree/main/examples/custom-node-components) - Custom transaction pool
|
||||
- [custom-payload-builder](https://github.com/paradigmxyz/reth/tree/main/examples/custom-payload-builder) - Custom payload builder
|
||||
- [custom-evm](https://github.com/paradigmxyz/reth/tree/main/examples/custom-evm) - Custom EVM configuration
|
||||
|
||||
## Component Lifecycle
|
||||
|
||||
Components follow a specific lifecycle starting from node builder initialization to shutdown:
|
||||
|
||||
Reference in New Issue
Block a user