diff --git a/Cargo.lock b/Cargo.lock index fe52b482d..d0dfbe036 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,18 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "TinyUFO" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "14cc2a7d25b49809b7bf199e443f949c4078ff48a492e8fb4cbc0477bdcf7bc9" +dependencies = [ + "ahash", + "crossbeam-queue", + "crossbeam-skiplist", + "flurry", +] + [[package]] name = "actix-codec" version = "0.5.2" @@ -176,6 +188,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75" dependencies = [ "cfg-if", + "const-random", "getrandom 0.3.3", "once_cell", "version_check", @@ -1657,6 +1670,26 @@ version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" +[[package]] +name = "const-random" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "87e00182fe74b066627d63b85fd550ac2998d4b0bd86bfed477a0ae4c7c71359" +dependencies = [ + "const-random-macro", +] + +[[package]] +name = "const-random-macro" +version = "0.1.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9d839f2a20b0aee515dc581a6172f2321f96cab76c1a38a4c584a194955390e" +dependencies = [ + "getrandom 0.2.16", + "once_cell", + "tiny-keccak", +] + [[package]] name = "const_format" version = "0.2.35" @@ -1760,6 +1793,25 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "crossbeam-queue" +version = "0.3.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "803d13fb3b09d88be9f4dbc29062c66b19bf7170867ceb746d2a8689bf6c7a26" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-skiplist" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df29de440c58ca2cc6e587ec3d22347551a32435fbde9d2bff64e78a9ffa151b" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + [[package]] name = "crossbeam-utils" version = "0.8.21" @@ -2396,6 +2448,18 @@ dependencies = [ "rustc_version 0.2.3", ] +[[package]] +name = "flurry" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf5efcf77a4da27927d3ab0509dec5b0954bb3bc59da5a1de9e52642ebd4cdf9" +dependencies = [ + "ahash", + "num_cpus", + "parking_lot", + "seize", +] + [[package]] name = "fnv" version = "1.0.7" @@ -4608,18 +4672,6 @@ dependencies = [ "memchr", ] -[[package]] -name = "quick_cache" -version = "0.6.22" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d1c821816e9b928e20e92ed59bb3ac4aab321d16ca2316871c9fe7ca739cd477" -dependencies = [ - "ahash", - "equivalent", - "hashbrown 0.16.1", - "parking_lot", -] - [[package]] name = "quinn" version = "0.11.8" @@ -5663,6 +5715,12 @@ dependencies = [ "libc", ] +[[package]] +name = "seize" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "689224d06523904ebcc9b482c6a3f4f7fb396096645c4cd10c0d2ff7371a34d3" + [[package]] name = "semver" version = "0.9.0" @@ -6184,6 +6242,7 @@ checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" name = "stratus" version = "0.20.1" dependencies = [ + "TinyUFO", "alloy-consensus", "alloy-dyn-abi", "alloy-eips", @@ -6209,6 +6268,7 @@ dependencies = [ "display_json", "dotenvy", "fake", + "foldhash 0.2.0", "futures", "futures-channel", "futures-timer", @@ -6241,7 +6301,6 @@ dependencies = [ "phf 0.13.1", "phf_codegen", "pin-project", - "quick_cache", "rand 0.9.4", "rdkafka", "reqwest 0.12.24", @@ -6570,6 +6629,15 @@ dependencies = [ "time-core", ] +[[package]] +name = "tiny-keccak" +version = "2.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c9d3793400a45f954c52e73d068316d76b6f4e36977e3fcebb13a2721e80237" +dependencies = [ + "crunchy", +] + [[package]] name = "tinystr" version = "0.8.1" diff --git a/Cargo.toml b/Cargo.toml index e19deca79..275650a1c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,6 +18,7 @@ const_format = "=0.2.35" const-hex = "=1.19.0" derive_more = "=2.0.1" derive-new = "=0.7.0" +foldhash = "=0.2.0" hash_hasher = "=2.0.4" hex_fmt = "=0.3.0" hex-literal = "=1.1.0" @@ -36,7 +37,7 @@ rustc-hash = "=2.1.1" smallvec = "=1.15.1" static_assertions = "=1.1.0" strum = { version = "=0.27.2", features = ["derive"] } -quick_cache = "=0.6.22" +TinyUFO = "=0.8.1" sugars = "=3.0.1" thiserror = "=2.0.18" uuid = { version = "=1.23.3", features = ["v7"]} diff --git a/src/eth/executor/evm/mod.rs b/src/eth/executor/evm/mod.rs index fc174804f..59f220533 100644 --- a/src/eth/executor/evm/mod.rs +++ b/src/eth/executor/evm/mod.rs @@ -128,7 +128,7 @@ impl Evm { .into(); // CREATE transactions need to be traced for blockscout to work correctly - if tx.result.deployed_contract_address.is_none() && trace_unsuccessful_only && matches!(tx.result.result, ExecutionResult::Success) { + if tx.output.deployed_contract_address.is_none() && trace_unsuccessful_only && matches!(tx.output.result, ExecutionResult::Success) { return Ok(default_trace(tracer_type, tx)); } @@ -137,10 +137,10 @@ impl Evm { .journaled_state .database .storage - .read_block(BlockFilter::Number(tx.evm_input.block_number))? + .read_block(BlockFilter::Number(tx.input.block_number))? .ok_or_else(|| { StratusError::Storage(StorageError::BlockNotFound { - filter: BlockFilter::Number(tx.evm_input.block_number), + filter: BlockFilter::Number(tx.input.block_number), }) })?; @@ -152,7 +152,7 @@ impl Evm { block_number: Some(block.number().as_u64()), base_fee: None, }; - let inspect_input: TransactionExecutionInput = tx.evm_input; + let inspect_input: TransactionExecutionInput = tx.input; let target = inspect_input.block_number.prev().unwrap_or_default(); self.evm.journaled_state.database.reset(ExecutionKind::CallPast(target)); @@ -166,7 +166,7 @@ impl Evm { if tx.info.hash == tx_hash { break; } - let tx_input: TransactionExecutionInput = tx.execution.evm_input; + let tx_input: TransactionExecutionInput = tx.execution.input; // Configure EVM state evm.fill_env(tx_input); diff --git a/src/eth/executor/evm/session.rs b/src/eth/executor/evm/session.rs index dab462b70..3d8401468 100644 --- a/src/eth/executor/evm/session.rs +++ b/src/eth/executor/evm/session.rs @@ -1,5 +1,6 @@ use std::sync::Arc; +use alloy_primitives::Uint; use anyhow::anyhow; use revm::Database; use revm::DatabaseRef; @@ -73,6 +74,11 @@ impl DatabaseRef for RevmSession { fn basic_ref(&self, address: revm::primitives::Address) -> Result, Self::Error> { // retrieve account let address: Address = address.into(); + + if address.is_ignored() { + return Ok(None); + } + let account = self.storage.read_account(address, self.kind)?; Ok(Some(account.into())) } @@ -80,6 +86,11 @@ impl DatabaseRef for RevmSession { fn storage_ref(&self, address: revm::primitives::Address, index: U256) -> Result { // convert slot let address: Address = address.into(); + + if address.is_ignored() { + return Ok(Uint::default()); + } + let index: SlotIndex = index.into(); // load slot from storage diff --git a/src/eth/executor/evm/types/input/call_execution.rs b/src/eth/executor/evm/types/input/call_execution.rs index 18cfba75f..907ad680c 100644 --- a/src/eth/executor/evm/types/input/call_execution.rs +++ b/src/eth/executor/evm/types/input/call_execution.rs @@ -8,13 +8,11 @@ use crate::eth::executor::evm::types::EvmInput; use crate::eth::executor::evm::types::GAS_MAX_LIMIT; use crate::eth::executor::evm::types::GeneralRevm; use crate::eth::storage::ExecutionKind; -use crate::eth::storage::TxCount; use crate::eth::types::Address; use crate::eth::types::Block; use crate::eth::types::BlockNumber; use crate::eth::types::Bytes; use crate::eth::types::CallInput; -use crate::eth::types::PendingBlockHeader; use crate::eth::types::PointInTime; use crate::eth::types::StratusError; use crate::eth::types::UnixTime; @@ -64,19 +62,6 @@ pub struct CallExecutionInput { } impl CallExecutionInput { - /// Creates from a call that was sent directly to Stratus with `eth_call` or `eth_estimateGas` for a pending block. - pub fn from_pending_block(input: CallInput, pending_header: PendingBlockHeader, tx_count: TxCount) -> Self { - Self { - from: input.from.unwrap_or(Address::ZERO), - to: input.to.map_into(), - value: input.value, - data: input.data, - block_number: pending_header.number, - block_timestamp: *pending_header.timestamp, - kind: ExecutionKind::CallPending(pending_header.number, tx_count), - } - } - /// Creates from a call that was sent directly to Stratus with `eth_call` or `eth_estimateGas` for a mined block. pub fn try_from_mined_block(input: CallInput, block: Block, point_in_time: PointInTime) -> anyhow::Result { let kind = match point_in_time { diff --git a/src/eth/executor/evm/types/mod.rs b/src/eth/executor/evm/types/mod.rs index 87a2dbbaf..543cbf507 100644 --- a/src/eth/executor/evm/types/mod.rs +++ b/src/eth/executor/evm/types/mod.rs @@ -22,6 +22,7 @@ pub use input::transaction_execution::TransactionExecutionInput; pub use output::access_list::AccessListOutput; pub use output::call_execution::CallExecutionOutput; pub use output::transaction_execution::TransactionExecutionOutput; +pub use output::transaction_execution::TransactionExecutionResult; /// Maximum gas limit allowed for a transaction. Prevents a transaction from consuming too many resources. #[cfg(feature = "dev")] diff --git a/src/eth/executor/evm/types/output/call_execution.rs b/src/eth/executor/evm/types/output/call_execution.rs index 57fb0a613..6e9020bf8 100644 --- a/src/eth/executor/evm/types/output/call_execution.rs +++ b/src/eth/executor/evm/types/output/call_execution.rs @@ -21,6 +21,10 @@ pub struct CallExecutionOutput { } impl CallExecutionOutput { + pub fn is_success(&self) -> bool { + self.success + } + fn parse_revm_result(result: RevmExecutionResult) -> (Bytes, Gas, bool) { match result { RevmExecutionResult::Success { output, gas, .. } => { diff --git a/src/eth/executor/evm/types/output/transaction_execution.rs b/src/eth/executor/evm/types/output/transaction_execution.rs index 5505486af..0d34e093b 100644 --- a/src/eth/executor/evm/types/output/transaction_execution.rs +++ b/src/eth/executor/evm/types/output/transaction_execution.rs @@ -1,30 +1,45 @@ use alloy_primitives::B256; +use derive_more::Deref; +use derive_more::DerefMut; use display_json::DebugAsJson; use hex_literal::hex; use itertools::Itertools; use revm::context::result::ExecutionResult as RevmExecutionResult; use revm_state::EvmState; -use crate::eth::executor::AccountChanges; -use crate::eth::executor::Changes; use crate::eth::executor::ExecutionResult; +use crate::eth::executor::State; use crate::eth::executor::evm::RevmResultAndState; +use crate::eth::executor::types::state::AccountChanges; +use crate::eth::executor::types::state::Complete; use crate::eth::types::Account; use crate::eth::types::Address; use crate::eth::types::Bytes; use crate::eth::types::ExternalReceipt; use crate::eth::types::Gas; use crate::eth::types::Log; -use crate::eth::types::Slot; use crate::eth::types::StratusError; use crate::eth::types::Wei; use crate::ext::not; use crate::log_and_err; /// Output of a transaction executed in the EVM. -#[derive(DebugAsJson, Clone, PartialEq, Eq, serde::Serialize, Default)] +#[derive(DebugAsJson, Clone, PartialEq, Eq, serde::Serialize, Default, Deref, DerefMut)] #[cfg_attr(test, derive(fake::Dummy))] pub struct TransactionExecutionOutput { + /// Status of the execution. + #[deref] + #[deref_mut] + pub outcome: TransactionExecutionResult, + + /// Storage changes that happened during the transaction execution. + pub state: State, +} + +/// Output of a transaction executed in the EVM. +#[derive(DebugAsJson, Clone, PartialEq, Eq, serde::Serialize, Default)] +#[cfg_attr(test, derive(fake::Dummy))] +pub struct TransactionExecutionResult { /// Status of the execution. pub result: ExecutionResult, @@ -37,9 +52,6 @@ pub struct TransactionExecutionOutput { /// Consumed gas. pub gas_used: Gas, - /// Storage changes that happened during the transaction execution. - pub changes: Changes, - /// The contract address if the executed transaction deploys a contract. pub deployed_contract_address: Option
, } @@ -61,17 +73,19 @@ impl TransactionExecutionOutput { let sender_next_nonce = sender_changes.nonce.next_nonce(); sender_changes.nonce.apply(sender_next_nonce); - let mut changes = Changes::default(); + let mut changes = State::default(); changes.accounts.insert(address, sender_changes); // crete execution and apply costs let mut execution = Self { - result: ExecutionResult::new_reverted("reverted externally".into()), // assume it reverted - output: Bytes::default(), // we cannot really know without performing an eth_call to the external system - logs: Vec::new(), - gas_used: Gas::from(receipt.gas_used), - changes, - deployed_contract_address: None, + outcome: TransactionExecutionResult { + result: ExecutionResult::new_reverted("reverted externally".into()), // assume it reverted + output: Bytes::default(), // we cannot really know without performing an eth_call to the external system + logs: Vec::new(), + gas_used: Gas::from(receipt.gas_used), + deployed_contract_address: None, + }, + state: changes, }; execution.apply_receipt(receipt)?; Ok(execution) @@ -182,7 +196,7 @@ impl TransactionExecutionOutput { if execution_cost > Wei::ZERO { // find sender changes let sender_address: Address = receipt.0.from.into(); - let Some(sender_changes) = self.changes.accounts.get_mut(&sender_address) else { + let Some(sender_changes) = self.state.accounts.get_mut(&sender_address) else { return log_and_err!("sender changes not present in execution when applying execution costs"); }; @@ -268,17 +282,18 @@ impl TransactionExecutionOutput { } } - fn parse_revm_state(revm_state: EvmState) -> Result<(Changes, Option
), StratusError> { + fn parse_revm_state(revm_state: EvmState) -> Result<(State, Option
), StratusError> { let mut deployed_contract_address = None; - let mut execution_changes = Changes::default(); - + let mut execution_changes = State::default(); + // might be improved by only keeping slots read from perm and modified slots + // and discard stots found in temp and cache for (revm_address, mut revm_account) in revm_state { let address: Address = revm_address.into(); + if address.is_ignored() { continue; } - // apply changes according to account status tracing::debug!( %address, status = ?revm_account.status, @@ -288,27 +303,15 @@ impl TransactionExecutionOutput { "evm account" ); - if !(revm_account.is_created() || revm_account.is_touched()) { - continue; - } - if revm_account.is_created() && revm_account.info.code.is_some() { deployed_contract_address = Some(address); } let storage = std::mem::take(&mut revm_account.storage); - let account_modified_slots: Vec = storage - .into_iter() - .filter_map(|(index, value)| match value.is_changed() { - true => Some(Slot::new(index.into(), value.present_value.into())), - false => None, - }) - .collect(); - - execution_changes.insert_slot_changes(address, account_modified_slots); - if revm_account.is_changed() { - execution_changes.insert_account_changes(address, revm_account.into()); - } + let account_slots = storage.into_iter().map(|(index, value)| (index.into(), value.into())).collect(); + + execution_changes.insert_slots(address, account_slots); + execution_changes.insert_account(address, revm_account.into()); } Ok((execution_changes, deployed_contract_address)) } @@ -323,12 +326,14 @@ impl TryFrom for TransactionExecutionOutput { tracing::debug!(?result, %gas, tx_output_len = %tx_output.len(), %tx_output, "evm executed"); Ok(TransactionExecutionOutput { - result, - output: tx_output, - logs, - gas_used: gas, - changes, - deployed_contract_address, + outcome: TransactionExecutionResult { + result, + output: tx_output, + logs, + gas_used: gas, + deployed_contract_address, + }, + state: changes, }) } } @@ -380,7 +385,7 @@ mod tests { assert_eq!(execution.gas_used, Gas::from(receipt.gas_used)); // Verify sender changes - let sender_changes = execution.changes.accounts.get(&sender_address).unwrap(); + let sender_changes = execution.state.accounts.get(&sender_address).unwrap(); // Nonce should be incremented let modified_nonce = *sender_changes.nonce.value(); @@ -628,13 +633,13 @@ mod tests { // Set up execution with sender account let mut sender_changes = AccountChanges::default(); sender_changes.apply_original(sender); - let mut accounts = HashMap::with_hasher(hash_hasher::HashBuildHasher::default()); + let mut accounts = HashMap::with_hasher(foldhash::fast::FixedState::default()); accounts.insert(sender_address, sender_changes); - let changes = Changes:: { + let changes = State:: { accounts, ..Default::default() }; - execution.changes = changes; + execution.state = changes; execution.gas_used = Gas::from(100u64); // Create a receipt with higher gas used and execution cost @@ -650,7 +655,7 @@ mod tests { execution.apply_receipt(&receipt).unwrap(); // Verify sender balance was reduced by execution cost - let sender_changes = execution.changes.accounts.get(&sender_address).unwrap(); + let sender_changes = execution.state.accounts.get(&sender_address).unwrap(); let modified_balance = *sender_changes.balance.value(); assert_eq!(modified_balance, Wei::from(900u64)); // 1000 - 100 } diff --git a/src/eth/executor/evm/util.rs b/src/eth/executor/evm/util.rs index c2fc2f12c..48360cacf 100644 --- a/src/eth/executor/evm/util.rs +++ b/src/eth/executor/evm/util.rs @@ -31,13 +31,13 @@ pub fn default_trace(tracer_type: GethDebugTracerType, tx: TransactionExecution) GethDebugTracerType::BuiltInTracer(GethDebugBuiltInTracerType::FourByteTracer) => FourByteFrame::default().into(), // HACK: Spoof empty call frame to prevent Blockscout from retrying unnecessary trace calls GethDebugTracerType::BuiltInTracer(GethDebugBuiltInTracerType::CallTracer) => { - let (typ, to) = match tx.evm_input.to { - Some(_) => ("CALL".to_string(), tx.evm_input.to.map_into()), - None => ("CREATE".to_string(), tx.result.deployed_contract_address.map_into()), + let (typ, to) = match tx.input.to { + Some(_) => ("CALL".to_string(), tx.input.to.map_into()), + None => ("CREATE".to_string(), tx.output.deployed_contract_address.map_into()), }; CallFrame { - from: tx.evm_input.from.into(), + from: tx.input.from.into(), to, typ, ..Default::default() diff --git a/src/eth/executor/mod.rs b/src/eth/executor/mod.rs index d5c022af1..8fda1aa15 100644 --- a/src/eth/executor/mod.rs +++ b/src/eth/executor/mod.rs @@ -1,7 +1,7 @@ mod config; mod evm; mod evm_worker_pool; -mod types; +pub mod types; use std::mem; use std::sync::Arc; @@ -18,22 +18,16 @@ pub use evm::types::EvmExecutionMetrics; pub use evm::types::EvmKind; pub use evm::types::TransactionExecutionInput; pub use evm::types::TransactionExecutionOutput; +pub use evm::types::TransactionExecutionResult; use parking_lot::Mutex; use tracing::Span; use tracing::debug_span; #[cfg(feature = "tracing")] use tracing::info_span; -pub use types::AccountChanges; -pub use types::AccountOriginalsReader; -pub use types::Changes; -pub use types::Complete; -pub use types::CompleteValue; pub use types::ExecutionResult; pub use types::ExecutorError; -pub use types::Incomplete; -pub use types::IncompleteValue; pub use types::RevertReason; -pub use types::Stage; +pub use types::State; pub use types::TransactionExecution; #[cfg(feature = "metrics")] @@ -178,11 +172,11 @@ impl Executor { tracing::info!(%block_number, tx_hash = %tx.hash(), "reexecuting external transaction"); let tx_input: TransactionInput = tx.try_into()?; - let (pending_block, _) = self.storage.read_pending_block_header(); + let pending_block = self.storage.read_pending_block_header(); let mut evm_input = TransactionExecutionInput::from_eth_transaction(&tx_input, pending_block.number, *pending_block.timestamp); // when transaction externally failed, create fake transaction instead of reexecuting - let tx_execution = match receipt.is_success() { + let (tx_execution, state) = match receipt.is_success() { // successful external transaction, re-execute locally true => { // re-execute transaction @@ -222,7 +216,10 @@ impl Executor { metrics::inc_executor_external_transaction_gas(evm_result.gas_used.as_u64() as usize, tx_contract, tx_function); } - TransactionExecution::new(tx_input.transaction_info, tx_input.signature, evm_input, evm_result) + ( + TransactionExecution::new(tx_input.transaction_info, tx_input.signature, evm_input, evm_result.outcome), + evm_result.state, + ) } // // failed external transaction, re-create from receipt without re-executing @@ -241,12 +238,15 @@ impl Executor { evm_input.gas_limit = tx_input.execution_info.gas_limit; evm_input.gas_price = tx_input.execution_info.gas_price; - TransactionExecution::new(tx_input.transaction_info, tx_input.signature, evm_input, evm_result) + ( + TransactionExecution::new(tx_input.transaction_info, tx_input.signature, evm_input, evm_result.outcome), + evm_result.state, + ) } }; // persist state - self.miner.save_execution(tx_execution)?; + self.miner.save_execution(tx_execution, state)?; Ok(()) } @@ -255,7 +255,11 @@ impl Executor { // ------------------------------------------------------------------------- /// Validates that the target account is a contract, reading it from storage at the given point in time. - pub fn validate_to_is_contract(&self, to_address: Address, kind: ExecutionKind) -> Result<(), StratusError> { + pub fn validate_to_is_contract(&self, to_address: Address, mut kind: ExecutionKind) -> Result<(), StratusError> { + // small warm up + if matches!(kind, ExecutionKind::Transaction) { + kind = ExecutionKind::RPC(PointInTime::Pending); + } let account = self.storage.read_account(to_address, kind)?; if account.bytecode.is_none() { if self.reject_not_contract { @@ -343,7 +347,7 @@ impl Executor { }); // prepare evm input - let (pending_header, _) = self.storage.read_pending_block_header(); + let pending_header = self.storage.read_pending_block_header(); let evm_input = TransactionExecutionInput::from_eth_transaction(&tx_input, pending_header.number, *pending_header.timestamp); // execute transaction in evm (retry only in case of conflict, but do not retry on other failures) @@ -359,26 +363,26 @@ impl Executor { "executing local transaction attempt" ); - let (evm_result, evm_metrics) = self.evms.execute(EvmRoute::Transaction(evm_input.clone()))?; // this clone can be avoided i think + let (evm_result, evm_metrics): (TransactionExecutionOutput, EvmExecutionMetrics) = self.evms.execute(EvmRoute::Transaction(evm_input.clone()))?; // save execution to temporary storage // in case of failure, retry if conflict or abandon if unexpected error - let tx_execution = TransactionExecution::new(tx_input.transaction_info, tx_input.signature, evm_input, evm_result); + let tx_execution = TransactionExecution::new(tx_input.transaction_info, tx_input.signature, evm_input, evm_result.outcome); #[cfg(feature = "metrics")] - let gas_used = tx_execution.result.gas_used; + let gas_used = tx_execution.output.gas_used; #[cfg(feature = "metrics")] let function = codegen::function_sig(&tx_input.execution_info.input); #[cfg(feature = "metrics")] let contract = codegen::contract_name(&tx_input.execution_info.to); - if let ExecutionResult::Reverted { reason } = &tx_execution.result.result { + if let ExecutionResult::Reverted { reason } = &tx_execution.output.result { tracing::info!(?reason, "local transaction execution reverted"); #[cfg(feature = "metrics")] metrics::inc_executor_local_transaction_reverts(contract, function, reason.0.as_ref()); } - match self.miner.save_execution(tx_execution) { + match self.miner.save_execution(tx_execution, evm_result.state) { Ok(_) => { // track metrics #[cfg(feature = "metrics")] @@ -425,22 +429,16 @@ impl Executor { "executing read-only local transaction" ); - // execute - let evm_input = match point_in_time { - PointInTime::Pending => { - let (pending_header, tx_count) = self.storage.read_pending_block_header(); - CallExecutionInput::from_pending_block(call_input.clone(), pending_header, tx_count) - } - point_in_time => { - // NOTE: this read is way more expensive that what we theoretically need, we only need to get the timestamp - // and block number, however this is not possible in the current rocksdb configuration. - let Some(block) = self.storage.read_block(point_in_time.into())? else { - return Err(RpcError::BlockFilterInvalid { filter: point_in_time.into() }.into()); - }; - CallExecutionInput::try_from_mined_block(call_input.clone(), block, point_in_time)? - } + let Some(block) = self.storage.read_block(point_in_time.into())? else { + return Err(RpcError::BlockFilterInvalid { filter: point_in_time.into() }.into()); }; + #[cfg(feature = "metrics")] + let (function, contract) = { (codegen::function_sig(&call_input.data), codegen::contract_name(&call_input.to)) }; + + // execute + let evm_input = CallExecutionInput::try_from_mined_block(call_input, block, point_in_time)?; + let evm_route = match point_in_time { PointInTime::Pending | PointInTime::Latest => EvmRoute::CallPresent(evm_input), PointInTime::Past(_) => EvmRoute::CallPast(evm_input), @@ -450,9 +448,6 @@ impl Executor { // track metrics #[cfg(feature = "metrics")] { - let function = codegen::function_sig(&call_input.data); - let contract = codegen::contract_name(&call_input.to); - match &evm_result { Ok((_, evm_metrics)) => { metrics::inc_executor_local_call(start.elapsed(), true, contract, function); diff --git a/src/eth/executor/types/mod.rs b/src/eth/executor/types/mod.rs index 6bae5a47d..e8ff7e59d 100644 --- a/src/eth/executor/types/mod.rs +++ b/src/eth/executor/types/mod.rs @@ -1,20 +1,13 @@ -mod changes; mod error; mod execution_result; +pub mod state; mod task; mod transaction_execution; -pub use changes::AccountChanges; -pub use changes::AccountOriginalsReader; -pub use changes::Changes; -pub use changes::Complete; -pub use changes::CompleteValue; -pub use changes::Incomplete; -pub use changes::IncompleteValue; -pub use changes::Stage; pub use error::ExecutorError; pub use execution_result::ExecutionResult; pub use execution_result::RevertReason; +pub use state::State; pub use task::EvmRoute; pub use task::EvmTask; pub use task::ExecutionTask; diff --git a/src/eth/executor/types/state/mod.rs b/src/eth/executor/types/state/mod.rs new file mode 100644 index 000000000..37816524b --- /dev/null +++ b/src/eth/executor/types/state/mod.rs @@ -0,0 +1,194 @@ +pub mod values; + +use std::collections::HashMap; +use std::collections::hash_map::Entry; +use std::fmt::Debug; + +use foldhash::fast::FixedState; +use serde_with::serde_as; +pub use values::AccountChanges; +pub use values::Change; +pub use values::CompleteValue; +pub use values::IncompleteValue; + +use crate::eth::storage::permanent::rocks::types::BlockChangesRocksdb; +use crate::eth::types::Account; +use crate::eth::types::Address; +use crate::eth::types::SlotIndex; +use crate::eth::types::SlotValue; +use crate::ext::OptionExt; + +/// Stage marker: changes may be incomplete +#[derive(Clone, PartialEq, Eq, Default, serde::Serialize)] +pub struct Incomplete; + +/// Stage marker: every value is final (either changed by the block or filled with the original +/// account from perm). Safe to consume. +#[derive(Clone, serde::Serialize, PartialEq, Eq, Default)] +pub struct Final; + +/// Stage marker: may contain unchanged values +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, serde::Serialize)] +pub struct Complete; + +pub trait Stage: PartialEq + Eq + Default + Clone + serde::Serialize { + type AccountChangeField: Change; + type SlotChangeField: Change; +} + +impl Stage for Incomplete { + type AccountChangeField = IncompleteValue; + type SlotChangeField = SlotValue; +} + +impl Stage for Final { + type AccountChangeField = CompleteValue; + type SlotChangeField = SlotValue; +} + +impl Stage for Complete { + type AccountChangeField = CompleteValue; + type SlotChangeField = CompleteValue; +} + +#[serde_as] +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, Default)] +pub struct State { + pub accounts: HashMap, FixedState>, + #[serde_as(as = "Vec<(_, _)>")] + pub slots: HashMap<(Address, SlotIndex), S::SlotChangeField, FixedState>, +} + +#[cfg(test)] +impl fake::Dummy for State { + fn dummy_with_rng(faker: &fake::Faker, rng: &mut R) -> Self { + Self { + accounts: fake::Dummy::dummy_with_rng(faker, rng), + slots: fake::Dummy::dummy_with_rng(faker, rng), + } + } +} + +/// Creates the INCOMPLETE account changes. +impl From for State { + fn from(value: BlockChangesRocksdb) -> Self { + let accounts = value + .account_changes + .into_iter() + .map(|(address, changes)| { + ( + address.into(), + AccountChanges { + nonce: changes.nonce.into(), + balance: changes.balance.into(), + bytecode: changes.bytecode.map(|inner| inner.map_into()).into(), + }, + ) + }) + .collect(); + let slots = value + .slot_changes + .into_iter() + .map(|((addr, idx), value)| ((addr.into(), idx.into()), value.into())) + .collect(); + + Self { accounts, slots } + } +} + +/// Reads the original state of accounts from permanent storage. +pub trait AccountOriginalsReader { + /// Returns the original accounts for the given addresses. + fn read_accounts(&self, addresses: Vec
) -> anyhow::Result>; +} + +impl State { + pub fn insert_account(&mut self, address: Address, incoming_changes: AccountChanges) { + match self.accounts.entry(address) { + std::collections::hash_map::Entry::Occupied(mut entry) => { + let existing_changes = entry.get_mut(); + existing_changes.merge(incoming_changes); + } + std::collections::hash_map::Entry::Vacant(entry) => { + entry.insert(incoming_changes); + } + } + } + + pub fn insert_slot(&mut self, address: Address, index: SlotIndex, incoming_change: CompleteValue) { + match self.slots.entry((address, index)) { + Entry::Occupied(mut entry) => entry.get_mut().merge(incoming_change), + Entry::Vacant(entry) => { + entry.insert(incoming_change); + } + } + } + + pub fn insert_slots(&mut self, address: Address, slots: Vec<(SlotIndex, CompleteValue)>) { + slots.into_iter().for_each(|(index, change)| self.insert_slot(address, index, change)); + } + + pub fn merge(&mut self, other: State) { + other.accounts.into_iter().for_each(|(address, changes)| self.insert_account(address, changes)); + other + .slots + .into_iter() + .for_each(|((address, index), changes)| self.insert_slot(address, index, changes)); + } + + pub fn finalize(self) -> State { + let accounts = self + .accounts + .into_iter() + .filter_map(|(address, account)| { + if address.is_ignored() { + None + } else { + account.complete().map(|acc| (address, acc)) + } + }) + .collect(); + + let slots = self + .slots + .into_iter() + .filter_map(|((address, index), slot_value)| { + if address.is_ignored() { + None + } else { + slot_value.changed().map(|value| ((address, index), value)) + } + }) + .collect(); + State { accounts, slots } + } +} + +impl State { + /// Reads the original account state from `storage` and resolves every unset field, advancing to + /// [`Full`]. The only way to turn an `Incomplete` into `Full`. + /// + /// Accounts not present in permanent storage (newly created by the block) resolve to + /// `Account::default()`, which is their correct pre-state. + pub fn complete(self, storage: &impl AccountOriginalsReader) -> anyhow::Result> { + let addresses = self.accounts.keys().copied().collect::>(); + let original_accounts: HashMap = storage.read_accounts(addresses)?.into_iter().collect(); + + let accounts = self + .accounts + .into_iter() + .map(|(address, changes)| { + let original = original_accounts.get(&address).cloned().unwrap_or_default(); + (address, changes.complete(original)) + }) + .collect(); + + let slots = self + .slots + .into_iter() + .map(|(slot_key, slot_value)| (slot_key, CompleteValue::Changed(slot_value))) + .collect(); + + Ok(State { accounts, slots }) + } +} diff --git a/src/eth/executor/types/changes.rs b/src/eth/executor/types/state/values.rs similarity index 50% rename from src/eth/executor/types/changes.rs rename to src/eth/executor/types/state/values.rs index 4631d20d3..516736e9e 100644 --- a/src/eth/executor/types/changes.rs +++ b/src/eth/executor/types/state/values.rs @@ -1,147 +1,103 @@ -use std::collections::HashMap; use std::fmt::Debug; use std::ops::Deref; -use serde_with::serde_as; +use revm_state::EvmStorageSlot; use crate::alias::RevmBytecode; -use crate::eth::storage::permanent::rocks::types::BlockChangesRocksdb; +use crate::eth::executor::types::state::Complete; +use crate::eth::executor::types::state::Final; +use crate::eth::executor::types::state::Incomplete; +use crate::eth::executor::types::state::Stage; use crate::eth::types::Account; use crate::eth::types::Address; use crate::eth::types::Nonce; -use crate::eth::types::Slot; -use crate::eth::types::SlotIndex; use crate::eth::types::SlotValue; use crate::eth::types::Wei; use crate::ext::InfallibleExt; -use crate::ext::OptionExt; -/// Stage marker: changes may be incomplete -#[derive(serde::Serialize)] -pub struct Incomplete; +pub trait Change: Clone + Debug + PartialEq + Eq + Default + serde::Serialize { + type Inner: Default; + fn is_changed(&self) -> bool; + fn take_value(self) -> Self::Inner; + fn changed(self) -> Option; + fn changed_ref(&self) -> Option<&Self::Inner>; +} -/// Stage marker: every value is final (either changed by the block or filled with the original -/// account from perm). Safe to consume. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, serde::Serialize)] -pub struct Complete; +impl Change for SlotValue { + type Inner = SlotValue; -pub trait Stage: serde::Serialize { - type ChangeField: Clone + Debug + PartialEq + Eq + Default + serde::Serialize; -} + fn take_value(self) -> Self::Inner { + self + } -impl Stage for Incomplete { - type ChangeField = IncompleteValue; -} + fn changed(self) -> Option { + Some(self) + } -impl Stage for Complete { - type ChangeField = CompleteValue; -} + fn changed_ref(&self) -> Option<&Self::Inner> { + Some(self) + } -#[serde_as] -#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, Default)] -pub struct Changes { - pub accounts: HashMap, hash_hasher::HashBuildHasher>, - #[serde_as(as = "Vec<(_, _)>")] - pub slots: HashMap<(Address, SlotIndex), SlotValue, hash_hasher::HashBuildHasher>, + fn is_changed(&self) -> bool { + true + } } -#[cfg(test)] -impl fake::Dummy for Changes { - fn dummy_with_rng(faker: &fake::Faker, rng: &mut R) -> Self { - Self { - accounts: fake::Dummy::dummy_with_rng(faker, rng), - slots: fake::Dummy::dummy_with_rng(faker, rng), +impl Change for CompleteValue { + type Inner = T; + + fn take_value(self) -> T { + match self { + Self::Changed(value) => value, + Self::Original(value) => value, } } -} -/// Creates the INCOMPLETE account changes. -impl From for Changes { - fn from(value: BlockChangesRocksdb) -> Self { - let accounts = value - .account_changes - .into_iter() - .map(|(address, changes)| { - ( - address.into(), - AccountChanges { - nonce: changes.nonce.into(), - balance: changes.balance.into(), - bytecode: changes.bytecode.map(|inner| inner.map_into()).into(), - }, - ) - }) - .collect(); - let slots = value - .slot_changes - .into_iter() - .map(|((addr, idx), value)| ((addr.into(), idx.into()), value.into())) - .collect(); - - Self { accounts, slots } + fn changed(self) -> Option { + match self { + Self::Changed(value) => Some(value), + Self::Original(_) => None, + } } -} -/// Reads the original state of accounts from permanent storage. -pub trait AccountOriginalsReader { - /// Returns the original accounts for the given addresses. - fn read_accounts(&self, addresses: Vec
) -> anyhow::Result>; -} + fn changed_ref(&self) -> Option<&T> { + match self { + Self::Changed(value) => Some(value), + Self::Original(_) => None, + } + } -impl Changes { - /// Reads the original account state from `storage` and resolves every unset field, advancing to - /// [`Complete`]. The only way to turn an `Incomplete` into `Complete`. - /// - /// Accounts not present in permanent storage (newly created by the block) resolve to - /// `Account::default()`, which is their correct pre-state. - pub fn complete(self, storage: &impl AccountOriginalsReader) -> anyhow::Result> { - let addresses = self.accounts.keys().copied().collect::>(); - let original_accounts: HashMap = storage.read_accounts(addresses)?.into_iter().collect(); - - let accounts = self - .accounts - .into_iter() - .map(|(address, changes)| { - let original = original_accounts.get(&address).cloned().unwrap_or_default(); - (address, changes.complete(original)) - }) - .collect(); - Ok(Changes { accounts, slots: self.slots }) + fn is_changed(&self) -> bool { + matches!(self, Self::Changed(_)) } } -impl Changes { - pub fn insert_account_changes(&mut self, address: Address, incoming_changes: AccountChanges) { - match self.accounts.entry(address) { - std::collections::hash_map::Entry::Occupied(mut entry) => { - let existing_changes = entry.get_mut(); - existing_changes.merge(incoming_changes); - } - std::collections::hash_map::Entry::Vacant(entry) => { - entry.insert(incoming_changes); - } +impl Change for IncompleteValue { + type Inner = T; + + fn take_value(self) -> T { + match self { + Self::Changed(value) => value, + Self::Unset => T::default(), } } - pub fn insert_slot_changes(&mut self, address: Address, slots: Vec) { - for slot in slots { - self.slots.insert((address, slot.index), slot.value); + fn changed(self) -> Option { + match self { + Self::Changed(value) => Some(value), + Self::Unset => None, } } - pub fn merge(&mut self, other: Changes) { - for (address, changes) in other.accounts { - match self.accounts.entry(address) { - std::collections::hash_map::Entry::Occupied(mut entry) => { - let current_changes = entry.get_mut(); - current_changes.merge(changes); - } - std::collections::hash_map::Entry::Vacant(entry) => { - entry.insert(changes); - } - } + fn changed_ref(&self) -> Option<&T> { + match self { + Self::Changed(value) => Some(value), + Self::Unset => None, } - self.slots.extend(other.slots); + } + + fn is_changed(&self) -> bool { + matches!(self, Self::Changed(_)) } } @@ -172,7 +128,7 @@ impl Deref for CompleteValue { } } -impl CompleteValue +impl CompleteValue where T: PartialEq + Eq + Default, { @@ -183,10 +139,10 @@ where } } - pub fn take_value(self) -> T { - match self { - Self::Changed(value) => value, - Self::Original(value) => value, + // A mege should never make a value state unchanged. + pub fn merge(&mut self, change: Self) { + if change.is_changed() { + *self = change; } } @@ -210,11 +166,6 @@ where } } - /// Returns whether the value has been changed. - pub fn is_changed(&self) -> bool { - matches!(self, Self::Changed(_)) - } - pub fn from_diff(original: T, current: T) -> Self { match original == current { true => Self::Original(current), @@ -223,6 +174,15 @@ where } } +impl From for CompleteValue { + fn from(value: EvmStorageSlot) -> Self { + match value.is_changed() { + true => Self::Changed(value.present_value.into()), + false => Self::Original(value.present_value.into()), + } + } +} + /// Incomplete-stage field: either the block changed this value, or the original is not yet known #[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize)] pub enum IncompleteValue { @@ -258,14 +218,29 @@ where /// Changes that happened to an account during a transaction. #[derive(Clone, PartialEq, Eq, serde::Serialize, Default)] -pub struct AccountChanges { - pub nonce: S::ChangeField, - pub balance: S::ChangeField, - pub bytecode: S::ChangeField>, +pub struct AccountChanges { + pub nonce: S::AccountChangeField, + pub balance: S::AccountChangeField, + pub bytecode: S::AccountChangeField>, +} + +impl AccountChanges { + pub fn is_changed(&self) -> bool { + self.nonce.is_changed() || self.balance.is_changed() || self.bytecode.is_changed() + } + + pub fn to_account(self, address: Address) -> Account { + Account { + address, + nonce: self.nonce.take_value(), + balance: self.balance.take_value(), + bytecode: self.bytecode.take_value(), + } + } } impl AccountChanges { - /// Fills every unset field with its real original value, advancing to [`Complete`]. + /// Fills every unset field with its real original value, advancing to [`Full`]. pub fn complete(self, original: Account) -> AccountChanges { AccountChanges { nonce: self.nonce.complete(original.nonce), @@ -276,23 +251,18 @@ impl AccountChanges { } impl AccountChanges { - /// Updates an existing account state with changes that happened during the transaction. - pub fn apply_modifications(&mut self, modified_account: Account) { - self.nonce.apply(modified_account.nonce); - self.balance.apply(modified_account.balance); - self.bytecode.apply(modified_account.bytecode); + /// Checks if account nonce, balance or bytecode were modified. + pub fn complete(self) -> Option> { + self.is_changed().then(|| AccountChanges { + nonce: self.nonce, + balance: self.balance, + bytecode: self.bytecode, + }) } - pub fn merge(&mut self, other: AccountChanges) { - if other.nonce.is_changed() { - self.nonce = other.nonce; - } - if other.balance.is_changed() { - self.balance = other.balance; - } - if other.bytecode.is_changed() { - self.bytecode = other.bytecode; - } + self.nonce.merge(other.nonce); + self.balance.merge(other.balance); + self.bytecode.merge(other.bytecode); } pub(crate) fn apply_original(&mut self, original_account: Account) { @@ -300,20 +270,6 @@ impl AccountChanges { self.balance.apply_original(original_account.balance); self.bytecode.apply_original(original_account.bytecode); } - - /// Checks if account nonce, balance or bytecode were modified. - pub fn is_modified(&self) -> bool { - self.nonce.is_changed() || self.balance.is_changed() || self.bytecode.is_changed() - } - - pub fn to_account(self, address: Address) -> Account { - Account { - address, - nonce: self.nonce.take_value(), - balance: self.balance.take_value(), - bytecode: self.bytecode.take_value(), - } - } } impl From<(Address, AccountChanges)> for Account { diff --git a/src/eth/executor/types/transaction_execution.rs b/src/eth/executor/types/transaction_execution.rs index 16ac37ad8..ff3af332f 100644 --- a/src/eth/executor/types/transaction_execution.rs +++ b/src/eth/executor/types/transaction_execution.rs @@ -10,7 +10,7 @@ use crate::alias::AlloyLogPrimitive; use crate::alias::AlloyReceipt; use crate::alias::AlloyTransaction; use crate::eth::executor::TransactionExecutionInput; -use crate::eth::executor::TransactionExecutionOutput; +use crate::eth::executor::TransactionExecutionResult; use crate::eth::types::Log; use crate::eth::types::LogsBloom; use crate::eth::types::MinedData; @@ -25,8 +25,8 @@ use crate::ext::RuintExt; pub struct TransactionExecution { pub info: TransactionInfo, pub signature: Signature, - pub evm_input: TransactionExecutionInput, - pub result: TransactionExecutionOutput, + pub input: TransactionExecutionInput, + pub output: TransactionExecutionResult, } impl TransactionExecution { @@ -39,8 +39,8 @@ impl TransactionExecution { data: AlloyLogData::new_unchecked(log.topics_non_empty().into_iter().map(Into::into).collect(), log.data.clone().into()), }, block_hash: None, - block_number: Some(self.evm_input.block_number.as_u64()), - block_timestamp: Some(*self.evm_input.block_timestamp), + block_number: Some(self.input.block_number.as_u64()), + block_timestamp: Some(*self.input.block_timestamp), transaction_hash: Some(self.info.hash.into()), transaction_index: None, log_index: None, @@ -52,14 +52,14 @@ impl TransactionExecution { /// Computes the bloom filter from execution logs. fn compute_bloom(&self) -> LogsBloom { let mut bloom = LogsBloom::default(); - for log in self.result.logs.iter() { + for log in self.output.logs.iter() { bloom.accrue_log(log); } bloom } pub fn logs(&self) -> &Vec { - &self.result.logs + &self.output.logs } /// Builds an [`AlloyReceipt`] from this execution, optionally enriching it @@ -69,8 +69,8 @@ impl TransactionExecution { /// [`crate::eth::types::TransactionMined`] (with mined data). pub fn to_alloy_receipt(&self, alloy_logs: Vec, mined_data: Option) -> AlloyReceipt { let receipt = Receipt { - status: Eip658Value::Eip658(self.result.is_success()), - cumulative_gas_used: self.result.gas_used.into(), // TODO: implement cumulative gas used correctly + status: Eip658Value::Eip658(self.output.result.is_success()), + cumulative_gas_used: self.output.gas_used.into(), // TODO: implement cumulative gas used correctly logs: alloy_logs, }; @@ -92,14 +92,14 @@ impl TransactionExecution { transaction_hash: self.info.hash.into(), transaction_index: mined_data.map(|data| data.index.into()), block_hash: mined_data.map(|data| data.block_hash.into()), - block_number: Some(self.evm_input.block_number.as_u64()), - gas_used: self.result.gas_used.into(), - effective_gas_price: self.evm_input.gas_price, + block_number: Some(self.input.block_number.as_u64()), + gas_used: self.output.gas_used.into(), + effective_gas_price: self.input.gas_price, blob_gas_used: None, blob_gas_price: None, - from: self.evm_input.from.into(), - to: self.evm_input.to.map_into(), - contract_address: self.result.deployed_contract_address.map_into(), + from: self.input.from.into(), + to: self.input.to.map_into(), + contract_address: self.output.deployed_contract_address.map_into(), } } } @@ -115,7 +115,7 @@ impl From for TransactionInput { fn from(value: TransactionExecution) -> Self { Self { transaction_info: value.info, - execution_info: value.evm_input.into(), + execution_info: value.input.into(), signature: value.signature, } } diff --git a/src/eth/follower/importer/fetchers/block_with_changes.rs b/src/eth/follower/importer/fetchers/block_with_changes.rs index 4a3d7b1ac..c4ec560a9 100644 --- a/src/eth/follower/importer/fetchers/block_with_changes.rs +++ b/src/eth/follower/importer/fetchers/block_with_changes.rs @@ -2,8 +2,8 @@ use std::sync::Arc; use async_trait::async_trait; -use crate::eth::executor::Changes; -use crate::eth::executor::Incomplete; +use crate::eth::executor::State; +use crate::eth::executor::types::state::Incomplete; use crate::eth::follower::importer::fetch_with_retry; use crate::eth::follower::importer::fetchers::DataFetcher; use crate::eth::rpc::BlockchainClient; @@ -21,7 +21,7 @@ impl DataFetcher for BlockWithChangesFetcher { type FetchedType = (BlockRocksdb, BlockChangesRocksdb); // If we complete the ExecutionChanges in the fetcher we risk completing with data that is altered by // a prior block. - type PostProcessType = (Block, Changes); + type PostProcessType = (Block, State); async fn fetch(&self, block_number: BlockNumber) -> Self::FetchedType { let fetch_fn = |bn| self.chain.fetch_block_with_changes(bn); diff --git a/src/eth/follower/importer/importers/fake_leader.rs b/src/eth/follower/importer/importers/fake_leader.rs index b43b9282d..4a6133295 100644 --- a/src/eth/follower/importer/importers/fake_leader.rs +++ b/src/eth/follower/importer/importers/fake_leader.rs @@ -4,7 +4,6 @@ use anyhow::bail; use async_trait::async_trait; use crate::GlobalState; -use crate::eth::executor::Changes; use crate::eth::executor::Executor; use crate::eth::executor::ExecutorError; use crate::eth::follower::importer::fetchers::DataFetcher; @@ -13,9 +12,7 @@ use crate::eth::follower::importer::importers::ImportData; use crate::eth::follower::importer::importers::ImporterWorker; use crate::eth::miner::Miner; use crate::eth::miner::miner::interval_miner::commit_retry; -use crate::eth::miner::miner::interval_miner::mine_local_retry; use crate::eth::storage::StratusStorage; -use crate::eth::types::Block; use crate::eth::types::StratusError; pub struct FakeLeaderWorker { @@ -52,25 +49,19 @@ impl ImporterWorker for FakeLeaderWorker { } } } - let (mined_block, changes, miner_guard) = mine_local_retry(&self.miner); - let completed_expected_changes = expected_changes.complete(self.storage.as_ref())?; - if changes != completed_expected_changes { - tracing::error!( - ?changes, - ?completed_expected_changes, - "execution changes result mismatch between leader and fake leader" - ); + let miner_guard = self.miner.locks.mine_and_commit.lock(); + let (mined_block, changes) = self.miner.mine_local(); + + let final_expected_changes = expected_changes.complete(self.storage.as_ref())?.finalize(); + let final_changes = changes.clone().finalize(); + if final_changes != final_expected_changes { + tracing::error!(?mined_block, "execution changes result mismatch between leader and fake leader"); bail!("execution changes mismatch between leader and fake leader") } - // `expected_block` is built from `BlockRocksdb` (replicated), which drops per-tx - // `changes` and `metrics` (replaced with `Default::default()` on the way back). Build a - // normalized copy of the locally-mined block so the comparison checks fields that actually - // survive replication, leaving the original untouched for commit. - let normalized_mined_block = normalize_for_replication_compare(&mined_block); - if normalized_mined_block != expected_block { - tracing::error!(?normalized_mined_block, ?expected_block, "block mismatch between leader and fake leader"); + if mined_block != expected_block { + tracing::error!(?mined_block, ?expected_block, "block mismatch between leader and fake leader"); bail!("block mismatch between leader and fake leader") } @@ -78,14 +69,3 @@ impl ImporterWorker for FakeLeaderWorker { Ok(block_tx_len) } } - -/// Builds a copy of `block` with per-tx fields that do not survive `Block -> BlockRocksdb -> Block` -/// replication reset to their defaults, so it can be compared against the leader's replicated -/// block for equivalence. Does not mutate the input. -fn normalize_for_replication_compare(block: &Block) -> Block { - let mut normalized = block.clone(); - for tx in &mut normalized.transactions { - tx.execution.result.changes = Changes::default(); - } - normalized -} diff --git a/src/eth/follower/importer/importers/replication.rs b/src/eth/follower/importer/importers/replication.rs index d3aa82931..42b6e3266 100644 --- a/src/eth/follower/importer/importers/replication.rs +++ b/src/eth/follower/importer/importers/replication.rs @@ -2,8 +2,8 @@ use std::sync::Arc; use async_trait::async_trait; -use crate::eth::executor::Changes; -use crate::eth::executor::Incomplete; +use crate::eth::executor::State; +use crate::eth::executor::types::state::Incomplete; use crate::eth::follower::importer::importers::ImportData; use crate::eth::follower::importer::importers::ImporterWorker; use crate::eth::follower::importer::send_block_to_kafka; @@ -27,7 +27,7 @@ impl ImportData for ::DataType { #[async_trait] impl ImporterWorker for ReplicationWorker { - type DataType = (Block, Changes); + type DataType = (Block, State); async fn import(&self, (block, changes): Self::DataType) -> anyhow::Result { tracing::info!(block_number = %block.number(), "received block with changes"); diff --git a/src/eth/follower/importer/mod.rs b/src/eth/follower/importer/mod.rs index 1c40c4162..86fb2fe4f 100644 --- a/src/eth/follower/importer/mod.rs +++ b/src/eth/follower/importer/mod.rs @@ -270,13 +270,14 @@ mod tests { use hash_hasher::HashBuildHasher; - use crate::eth::executor::AccountChanges; - use crate::eth::executor::Changes; - use crate::eth::executor::CompleteValue; use crate::eth::executor::ExecutionResult; + use crate::eth::executor::State; use crate::eth::executor::TransactionExecution; use crate::eth::executor::TransactionExecutionInput; - use crate::eth::executor::TransactionExecutionOutput; + use crate::eth::executor::TransactionExecutionResult; + use crate::eth::executor::types::state::AccountChanges; + use crate::eth::executor::types::state::Complete; + use crate::eth::executor::types::state::CompleteValue; use crate::eth::follower::importer::fetchers::DataFetcher; use crate::eth::follower::importer::fetchers::block_with_changes::BlockWithChangesFetcher; use crate::eth::follower::importer::importers::ImporterWorker; @@ -301,7 +302,7 @@ mod tests { use crate::eth::types::UnixTime; use crate::eth::types::Wei; - impl AccountChanges { + impl AccountChanges { pub fn from_changed(account: Account) -> Self { Self { nonce: CompleteValue::Changed(account.nonce), @@ -312,26 +313,25 @@ mod tests { } /// Mines a block applying `changes` (mirrors the helper in `stratus_storage` tests). - fn mine_block(storage: &StratusStorage, changes: Changes) { - let (header, _) = storage.read_pending_block_header(); + fn mine_block(storage: &StratusStorage, state: State) { + let header = storage.read_pending_block_header(); let evm_input = TransactionExecutionInput::from_eth_transaction(&TransactionInput::default(), header.number, *header.timestamp); - let result = TransactionExecutionOutput { + let result = TransactionExecutionResult { result: ExecutionResult::Success, - changes, ..Default::default() }; let tx = TransactionExecution::new(TransactionInfo::default(), Signature::default(), evm_input, result); - storage.save_execution(tx).expect("save execution"); + storage.save_execution(tx, state).expect("save execution"); - let (block, block_changes) = storage.finish_pending_block().expect("finish pending block"); + let (block, block_changes) = storage.finish_pending_block(); storage.save_block(block.into(), block_changes).expect("save block"); } /// Builds `ExecutionChanges` that set `address`'s balance to `balance` (nonce/bytecode untouched). - fn balance_changes(address: Address, balance: Wei) -> Changes { - let mut changes = Changes::default(); + fn balance_changes(address: Address, balance: Wei) -> State { + let mut changes = State::default(); changes .accounts .insert(address, AccountChanges::from_changed(Account::new_with_balance(address, balance))); diff --git a/src/eth/miner/miner.rs b/src/eth/miner/miner.rs index b21b38bb6..a3b7c727d 100644 --- a/src/eth/miner/miner.rs +++ b/src/eth/miner/miner.rs @@ -13,8 +13,9 @@ use tokio::task::JoinSet; use tokio_util::sync::CancellationToken; use tracing::Span; -use crate::eth::executor::Changes; +use crate::eth::executor::State; use crate::eth::executor::TransactionExecution; +use crate::eth::executor::types::state::Complete; use crate::eth::miner::MinerMode; use crate::eth::storage::StorageError; use crate::eth::storage::StratusStorage; @@ -196,7 +197,7 @@ impl Miner { } /// Persists a transaction execution. - pub fn save_execution(&self, tx_execution: TransactionExecution) -> Result<(), StratusError> { + pub fn save_execution(&self, tx_execution: TransactionExecution, state: State) -> Result<(), StratusError> { let tx_hash = tx_execution.info.hash; // track @@ -210,7 +211,7 @@ impl Miner { let _save_execution_lock = if is_automine { Some(self.locks.save_execution.lock()) } else { None }; // save execution to temporary storage - self.storage.save_execution(tx_execution)?; + self.storage.save_execution(tx_execution, state)?; // notify if self.has_pending_tx_subscribers() { @@ -228,7 +229,7 @@ impl Miner { /// Mines external block and external transactions. /// /// Local transactions are not allowed to be part of the block. - pub fn mine_external(&self, external_block: ExternalBlock) -> anyhow::Result<(Block, Changes)> { + pub fn mine_external(&self, external_block: ExternalBlock) -> anyhow::Result<(Block, State)> { // track #[cfg(feature = "tracing")] let _span = info_span!("miner::mine_external", block_number = field::Empty).entered(); @@ -237,7 +238,7 @@ impl Miner { let _mine_lock = self.locks.mine.lock(); // mine block - let (pending_block, changes) = self.storage.finish_pending_block()?; + let (pending_block, changes) = self.storage.finish_pending_block(); let mut block: Block = pending_block.into(); Span::with(|s| s.rec_str("block_number", &block.header.number)); @@ -262,14 +263,14 @@ impl Miner { pub fn mine_local_and_commit(&self) -> anyhow::Result<(), StorageError> { let _mine_and_commit_lock = self.locks.mine_and_commit.lock(); - let (block, changes) = self.mine_local()?; + let (block, changes) = self.mine_local(); self.commit(CommitItem::Block(block), changes) } /// Mines local transactions. /// /// External transactions are not allowed to be part of the block. - pub fn mine_local(&self) -> anyhow::Result<(Block, Changes), StorageError> { + pub fn mine_local(&self) -> (Block, State) { #[cfg(feature = "tracing")] let _span = info_span!("miner::mine_local", block_number = field::Empty).entered(); @@ -277,25 +278,25 @@ impl Miner { let _mine_lock = self.locks.mine.lock(); // mine block - let (block, changes) = self.storage.finish_pending_block()?; + let (block, changes) = self.storage.finish_pending_block(); Span::with(|s| s.rec_str("block_number", &block.header.number)); - Ok((block.into(), changes)) + (block.into(), changes) } - pub fn commit(&self, item: CommitItem, changes: Changes) -> anyhow::Result<(), StorageError> { + pub fn commit(&self, item: CommitItem, changes: State) -> anyhow::Result<(), StorageError> { match item { CommitItem::Block(block) => self.commit_block(block, changes), CommitItem::ReplicationBlock(block) => { self.storage.set_pending_header(block.number(), block.timestamp()); - self.storage.finish_pending_block()?; + self.storage.finish_pending_block(); self.commit_block(block, changes) } } } /// Persists a mined block to permanent storage and prepares new block. - pub fn commit_block(&self, block: Block, changes: Changes) -> anyhow::Result<(), StorageError> { + pub fn commit_block(&self, block: Block, changes: State) -> anyhow::Result<(), StorageError> { let block_number = block.number(); // track @@ -382,7 +383,8 @@ pub mod interval_miner { use tokio::time::Instant; use tokio_util::sync::CancellationToken; - use crate::eth::executor::Changes; + use crate::eth::executor::State; + use crate::eth::executor::types::state::Complete; use crate::eth::miner::Miner; use crate::eth::miner::miner::CommitItem; use crate::eth::types::Block; @@ -414,25 +416,14 @@ pub mod interval_miner { // mine tracing::info!(lag_us = %tick.elapsed().as_micros(), "interval mining block"); - let (block, changes, miner_guard) = mine_local_retry(&miner); + let miner_guard = miner.locks.mine_and_commit.lock(); + let (block, changes) = miner.mine_local(); commit_retry(&miner, block, changes, miner_guard); } warn_task_rx_closed(TASK_NAME); } - pub fn mine_local_retry(miner: &Miner) -> (Block, Changes, MutexGuard<'_, ()>) { - let guard = miner.locks.mine_and_commit.lock(); - loop { - match miner.mine_local() { - Ok((block, changes)) => break (block, changes, guard), - Err(e) => { - tracing::error!(reason = ?e, "failed to mine block"); - } - } - } - } - - pub fn commit_retry(miner: &Miner, block: Block, changes: Changes, _miner_guard: MutexGuard<()>) { + pub fn commit_retry(miner: &Miner, block: Block, changes: State, _miner_guard: MutexGuard<()>) { loop { match miner.commit(CommitItem::Block(block.clone()), changes.clone()) { Ok(_) => break, diff --git a/src/eth/rpc/server.rs b/src/eth/rpc/server.rs index 5c5d476a3..9eff1ab32 100644 --- a/src/eth/rpc/server.rs +++ b/src/eth/rpc/server.rs @@ -50,7 +50,6 @@ use crate::eth::executor::AccessListOutput; use crate::eth::executor::CallExecutionOutput; use crate::eth::executor::Executor; use crate::eth::executor::ExecutorError; -use crate::eth::executor::TransactionExecutionOutput; use crate::eth::follower::ConsensusError; use crate::eth::follower::ImporterError; use crate::eth::follower::consensus::Consensus; @@ -1166,7 +1165,7 @@ fn eth_estimate_gas(params: Params<'_>, ctx: Arc, ext: Extensions) - } } -fn rpc_call(params: Params<'_>, ctx: Arc) -> Result { +fn rpc_call(params: Params<'_>, ctx: Arc) -> Result { // parse params let (params, call) = next_rpc_param::(params.sequence())?; let (_, filter) = next_rpc_param_or_default::(params)?; diff --git a/src/eth/rpc/types/log_filter_input.rs b/src/eth/rpc/types/log_filter_input.rs index fe8db108e..6b881a1ac 100644 --- a/src/eth/rpc/types/log_filter_input.rs +++ b/src/eth/rpc/types/log_filter_input.rs @@ -62,7 +62,7 @@ impl LogFilterInput { // translate point-in-time to block according to context let from = match from { - PointInTime::Pending => storage.read_pending_block_header().0.number, + PointInTime::Pending => storage.read_pending_block_header().number, PointInTime::Latest => storage.read_mined_block_number(), PointInTime::Past(number) => number, }; diff --git a/src/eth/storage/cache.rs b/src/eth/storage/cache.rs index 7f7c12f59..5c465c268 100644 --- a/src/eth/storage/cache.rs +++ b/src/eth/storage/cache.rs @@ -1,39 +1,35 @@ +use std::hash::DefaultHasher; use std::hash::Hash; -use std::time::Duration; +use std::hash::Hasher; +use std::sync::atomic::AtomicU64; +use std::sync::atomic::Ordering; use clap::Parser; use display_json::DebugAsJson; -use indexmap::Equivalent; -use quick_cache::UnitWeighter; -use quick_cache::sync::Cache; -use quick_cache::sync::DefaultLifecycle; -use quick_cache::sync::GuardResult; -use rustc_hash::FxBuildHasher; - -use crate::eth::executor::Changes; +use parking_lot::Mutex; +use tinyufo::TinyUfo; + +use crate::eth::executor::State; +use crate::eth::executor::types::state::Complete; use crate::eth::types::Account; use crate::eth::types::Address; use crate::eth::types::Slot; use crate::eth::types::SlotIndex; use crate::eth::types::SlotValue; +const INSERT_LOCK_SHARDS: usize = 64; + +type VersionedKey = (u64, K); + pub struct StorageCache { - slot_cache: Cache<(Address, SlotIndex), SlotValue, UnitWeighter, FxBuildHasher>, - account_cache: Cache, - account_latest_cache: Cache, - slot_latest_cache: Cache<(Address, SlotIndex), SlotValue, UnitWeighter, FxBuildHasher>, + account_latest_cache: TinyUfo, Account>, + slot_latest_cache: TinyUfo, SlotValue>, + generation: AtomicU64, + insert_locks: [Mutex<()>; INSERT_LOCK_SHARDS], } #[derive(DebugAsJson, Clone, Parser, serde::Serialize)] pub struct CacheConfig { - /// Capacity of slot cache - #[arg(long = "slot-cache-capacity", env = "SLOT_CACHE_CAPACITY", default_value = "100000")] - pub slot_cache_capacity: usize, - - /// Capacity of account cache - #[arg(long = "account-cache-capacity", env = "ACCOUNT_CACHE_CAPACITY", default_value = "20000")] - pub account_cache_capacity: usize, - /// Capacity of account history cache #[arg(long = "account-history-cache-capacity", env = "ACCOUNT_HISTORY_CACHE_CAPACITY", default_value = "20000")] pub account_history_cache_capacity: usize, @@ -52,125 +48,76 @@ impl CacheConfig { impl StorageCache { pub fn new(config: &CacheConfig) -> Self { Self { - slot_cache: Cache::with( - config.slot_cache_capacity, - config.slot_cache_capacity as u64, - UnitWeighter, - FxBuildHasher, - DefaultLifecycle::default(), - ), - account_cache: Cache::with( - config.account_cache_capacity, - config.account_cache_capacity as u64, - UnitWeighter, - FxBuildHasher, - DefaultLifecycle::default(), - ), - account_latest_cache: Cache::with( - config.account_history_cache_capacity, - config.account_history_cache_capacity as u64, - UnitWeighter, - FxBuildHasher, - DefaultLifecycle::default(), - ), - slot_latest_cache: Cache::with( - config.slot_history_cache_capacity, - config.slot_history_cache_capacity as u64, - UnitWeighter, - FxBuildHasher, - DefaultLifecycle::default(), - ), + account_latest_cache: TinyUfo::new(config.account_history_cache_capacity, config.account_history_cache_capacity), + slot_latest_cache: TinyUfo::new(config.slot_history_cache_capacity, config.slot_history_cache_capacity), + generation: AtomicU64::new(0), + insert_locks: std::array::from_fn(|_| Mutex::new(())), } } pub fn clear(&self) { - self.slot_cache.clear(); - self.account_cache.clear(); - self.account_latest_cache.clear(); - self.slot_latest_cache.clear(); + // TinyUFO does not expose a clear operation. Moving to a new key generation + // makes all existing entries inaccessible; they are reclaimed by normal eviction. + self.generation.fetch_add(1, Ordering::AcqRel); } - pub fn cache_slot_if_missing(&self, address: Address, slot: Slot) { - self.slot_cache.insert_if_missing((address, slot.index), slot.value); - } - - pub fn cache_account_if_missing(&self, account: Account) { - self.account_cache.insert_if_missing(account.address, account); - } - - fn _cache_account_and_slots_from_changes_impl( - changes: Changes, - account_cache: &Cache, - slot_cache: &Cache<(Address, SlotIndex), SlotValue, UnitWeighter, FxBuildHasher>, - ) { + fn cache_account_and_slots_from_changes_impl(&self, changes: &State) { // cache accounts - for (address, change) in changes.accounts { - let account = (address, change).into(); - account_cache.insert(address, account); + for (address, change) in changes.accounts.iter() { + let _guard = self.insert_lock(address); + let key = (self.generation(), *address); + let account = change.clone().to_account(*address); + let _ = self.account_latest_cache.put(key, account, 1); } // cache slots - for ((address, index), value) in changes.slots { - slot_cache.insert((address, index), value); + for ((address, index), value) in changes.slots.iter() { + let cache_key = (*address, *index); + let _guard = self.insert_lock(&cache_key); + let key = (self.generation(), cache_key); + let _ = self.slot_latest_cache.put(key, *value.value(), 1); } } - pub fn cache_account_and_slots_from_changes(&self, changes: Changes) { - Self::_cache_account_and_slots_from_changes_impl(changes, &self.account_cache, &self.slot_cache); - } - - pub fn cache_account_and_slots_latest_from_changes(&self, changes: Changes) { - Self::_cache_account_and_slots_from_changes_impl(changes, &self.account_latest_cache, &self.slot_latest_cache); - } - - pub fn get_slot(&self, address: Address, index: SlotIndex) -> Option { - self.slot_cache.get(&(address, index)).map(|value| Slot { value, index }) - } - - pub fn get_account(&self, address: Address) -> Option { - self.account_cache.get(&address) + pub fn cache_account_and_slots_latest_from_changes(&self, changes: &State) { + self.cache_account_and_slots_from_changes_impl(changes); } pub fn cache_account_latest_if_missing(&self, address: Address, account: Account) { - self.account_latest_cache.insert_if_missing(address, account); + let _guard = self.insert_lock(&address); + let key = (self.generation(), address); + if self.account_latest_cache.get(&key).is_none() { + let _ = self.account_latest_cache.put(key, account, 1); + } } pub fn cache_slot_latest_if_missing(&self, address: Address, slot: Slot) { - self.slot_latest_cache.insert_if_missing((address, slot.index), slot.value); + let cache_key = (address, slot.index); + let _guard = self.insert_lock(&cache_key); + let key = (self.generation(), cache_key); + if self.slot_latest_cache.get(&key).is_none() { + let _ = self.slot_latest_cache.put(key, slot.value, 1); + } } pub fn get_account_latest(&self, address: Address) -> Option { - self.account_latest_cache.get(&address) + self.account_latest_cache.get(&(self.generation(), address)) } pub fn get_slot_latest(&self, address: Address, index: SlotIndex) -> Option { - self.slot_latest_cache.get(&(address, index)).map(|value| Slot { value, index }) + self.slot_latest_cache + .get(&(self.generation(), (address, index))) + .map(|value| Slot { value, index }) } -} -trait CacheExt { - fn insert_if_missing(&self, key: Key, val: Val); -} + fn generation(&self) -> u64 { + self.generation.load(Ordering::Acquire) + } -impl CacheExt for Cache -where - Key: Hash + Equivalent + ToOwned + std::cmp::Eq, - Val: Clone, - We: quick_cache::Weighter + Clone, - B: std::hash::BuildHasher + Clone, - L: quick_cache::Lifecycle + Clone, -{ - fn insert_if_missing(&self, key: Key, val: Val) { - // None means wait forever, if someone else has the guard it will block. - // Some(Duration::ZERO) means "if someone has the guard return immediately" - // since we're only inserting the value if it is not cached yet the latter - // is the desired behavior. - match self.get_value_or_guard(&key, Some(Duration::ZERO)) { - GuardResult::Value(_) | GuardResult::Timeout => (), - GuardResult::Guard(g) => { - // this fails if an unguarded insert already inserted to this key - let _ = g.insert(val); - } - } + fn insert_lock(&self, key: &Key) -> parking_lot::MutexGuard<'_, ()> { + let mut hasher = DefaultHasher::new(); + key.hash(&mut hasher); + let shard = hasher.finish() as usize % INSERT_LOCK_SHARDS; + self.insert_locks[shard].lock() } } diff --git a/src/eth/storage/mod.rs b/src/eth/storage/mod.rs index 7f48e25e8..6822550ed 100644 --- a/src/eth/storage/mod.rs +++ b/src/eth/storage/mod.rs @@ -25,7 +25,6 @@ pub use temporary::compute_pending_block_number; pub use crate::eth::types::ExecutionKind; use crate::eth::types::StratusError; -pub use crate::eth::types::TxCount; // ----------------------------------------------------------------------------- // Config diff --git a/src/eth/storage/permanent/rocks/rocks_permanent.rs b/src/eth/storage/permanent/rocks/rocks_permanent.rs index cb3b34c3d..6817156d6 100644 --- a/src/eth/storage/permanent/rocks/rocks_permanent.rs +++ b/src/eth/storage/permanent/rocks/rocks_permanent.rs @@ -10,7 +10,8 @@ use super::rocks_cf_cache_config::RocksCfCacheConfig; use super::rocks_state::RocksStorageState; use super::types::BlockRocksdb; use crate::GlobalState; -use crate::eth::executor::Changes; +use crate::eth::executor::State; +use crate::eth::executor::types::state::Final; use crate::eth::rpc::BlockFilter; use crate::eth::rpc::LogFilter; use crate::eth::storage::MinedPointInTime; @@ -211,7 +212,7 @@ impl RocksPermanentStorage { }) } - pub fn save_genesis_block(&self, block: Block, accounts: Vec, account_changes: Changes) -> anyhow::Result<(), StorageError> { + pub fn save_genesis_block(&self, block: Block, accounts: Vec, account_changes: State) -> anyhow::Result<(), StorageError> { #[cfg(feature = "rocks_metrics")] { self.state.export_metrics().map_err(|err| StorageError::RocksError { err }).inspect_err(|e| { @@ -227,7 +228,7 @@ impl RocksPermanentStorage { }) } - pub fn save_block(&self, block: Block, account_changes: Changes) -> anyhow::Result<(), StorageError> { + pub fn save_block(&self, block: Block, account_changes: State) -> anyhow::Result<(), StorageError> { #[cfg(feature = "rocks_metrics")] { self.state.export_metrics().map_err(|err| StorageError::RocksError { err }).inspect_err(|e| { diff --git a/src/eth/storage/permanent/rocks/rocks_state.rs b/src/eth/storage/permanent/rocks/rocks_state.rs index 142d05679..982f8e1ed 100644 --- a/src/eth/storage/permanent/rocks/rocks_state.rs +++ b/src/eth/storage/permanent/rocks/rocks_state.rs @@ -40,13 +40,13 @@ use super::types::HashRocksdb; use super::types::SlotIndexRocksdb; use super::types::SlotValueRocksdb; use super::types::UnixTimeRocksdb; -use crate::eth::executor::Changes; +use crate::eth::executor::State; +use crate::eth::executor::types::state::Final; use crate::eth::rpc::BlockFilter; use crate::eth::rpc::LogFilter; use crate::eth::storage::MinedPointInTime; use crate::eth::storage::permanent::rocks::SerializeDeserializeWithContext; use crate::eth::storage::permanent::rocks::cf_versions::CfBlockChangesValue; -use crate::eth::storage::permanent::rocks::types::AccountChangesRocksdb; use crate::eth::storage::permanent::rocks::types::BlockChangesRocksdb; use crate::eth::types::Account; use crate::eth::types::Address; @@ -63,7 +63,6 @@ use crate::eth::types::SlotIndex; use crate::eth::types::TransactionMined; #[cfg(feature = "dev")] use crate::eth::types::Wei; -use crate::ext::OptionExt; #[cfg(feature = "metrics")] use crate::infra::metrics; use crate::log_and_err; @@ -217,37 +216,22 @@ impl RocksStorageState { } /// Updates the in-memory state with changes from transaction execution - fn prepare_batch_with_execution_changes(&self, changes: Changes, block_number: BlockNumber, batch: &mut WriteBatch) -> Result<()> { + fn prepare_batch_with_execution_changes(&self, changes: State, block_number: BlockNumber, batch: &mut WriteBatch) -> Result<()> { let mut block_changes = BlockChangesRocksdb::with_capacity(changes.accounts.len()); let block_number = block_number.into(); for (address, change) in changes.accounts { - let address: AddressRocksdb = address.into(); - if change.is_modified() { - let mut account_change_entry = AccountChangesRocksdb::default(); - let mut account_info_entry = self.accounts.get(&address)?.unwrap_or(AccountRocksdb::default().into()); - - if change.nonce.is_changed() { - let nonce = (*change.nonce.value()).into(); - account_info_entry.nonce = nonce; - account_change_entry.nonce = Some(nonce); - } - if change.balance.is_changed() { - let balance = (*change.balance.value()).into(); - account_info_entry.balance = balance; - account_change_entry.balance = Some(balance); - } - if change.bytecode.is_changed() { - let bytecode = change.bytecode.value().clone().map_into(); - account_info_entry.bytecode = bytecode.clone(); - account_change_entry.bytecode = Some(bytecode); - } + let address_rocks: AddressRocksdb = address.into(); + let account_change_entry = (&change).into(); + let account_info_entry: CfAccountsValue = match self.accounts.get(&address_rocks)? { + Some(existing_account) => existing_account.into_inner().update(change).into(), + None => change.to_account(address).into(), + }; - self.accounts.prepare_batch_insertion([(address, account_info_entry.clone())], batch)?; - self.accounts_history - .prepare_batch_insertion([((address, block_number), account_info_entry.into_inner().into())], batch)?; - block_changes.account_changes.insert(address, account_change_entry); - } + self.accounts.prepare_batch_insertion([(address_rocks, account_info_entry.clone())], batch)?; + self.accounts_history + .prepare_batch_insertion([((address_rocks, block_number), account_info_entry.into_inner().into())], batch)?; + block_changes.account_changes.insert(address_rocks, account_change_entry); } for ((address, slot_index), slot_value) in changes.slots { @@ -458,12 +442,12 @@ impl RocksStorageState { self.write_in_batch_for_multiple_cfs(write_batch) } - pub fn save_genesis_block(&self, block: Block, accounts: Vec, account_changes: Changes) -> Result<()> { + pub fn save_genesis_block(&self, block: Block, accounts: Vec, account_changes: State) -> Result<()> { let mut batch = WriteBatch::default(); let mut txs_batch = vec![]; for transaction in block.transactions.iter().cloned() { - txs_batch.push((transaction.info.hash.into(), transaction.evm_input.block_number.into())); + txs_batch.push((transaction.info.hash.into(), transaction.input.block_number.into())); } self.transactions.prepare_batch_insertion(txs_batch, &mut batch)?; @@ -501,16 +485,16 @@ impl RocksStorageState { self.write_in_batch_for_multiple_cfs(batch) } - pub fn save_block(&self, block: Block, account_changes: Changes) -> Result<()> { + pub fn save_block(&self, block: Block, account_changes: State) -> Result<()> { let mut batch = WriteBatch::default(); self.prepare_block_insertion(block, account_changes, &mut batch)?; self.write_in_batch_for_multiple_cfs(batch) } - pub fn prepare_block_insertion(&self, block: Block, account_changes: Changes, batch: &mut WriteBatch) -> Result<()> { + pub fn prepare_block_insertion(&self, block: Block, account_changes: State, batch: &mut WriteBatch) -> Result<()> { let mut txs_batch = vec![]; for transaction in block.transactions.iter().cloned() { - txs_batch.push((transaction.info.hash.into(), transaction.evm_input.block_number.into())); + txs_batch.push((transaction.info.hash.into(), transaction.input.block_number.into())); } self.transactions.prepare_batch_insertion(txs_batch, batch)?; @@ -596,6 +580,7 @@ impl RocksStorageState { #[cfg(feature = "dev")] pub fn save_account_code(&self, address: Address, code: Bytes) -> Result<()> { use crate::alias::RevmBytecode; + use crate::ext::OptionExt; let mut batch = WriteBatch::default(); @@ -799,7 +784,8 @@ mod tests { use super::*; use crate::eth::executor::TransactionExecution; use crate::eth::executor::TransactionExecutionInput; - use crate::eth::executor::TransactionExecutionOutput; + use crate::eth::executor::TransactionExecutionResult; + use crate::eth::executor::types::state::Complete; use crate::eth::types::BlockHeader; #[test] @@ -860,13 +846,13 @@ mod tests { }, transactions: vec![TransactionMined { execution: TransactionExecution { - evm_input: TransactionExecutionInput { + input: TransactionExecutionInput { block_number: number.into(), ..Faker.fake() }, - result: TransactionExecutionOutput { + output: TransactionExecutionResult { logs: vec![Faker.fake(), Faker.fake()], - ..Faker.fake() + ..Default::default() }, ..Faker.fake() }, @@ -874,7 +860,7 @@ mod tests { }], }; - state.save_block(block, Changes::default()).unwrap(); + state.save_block(block, State::::default().finalize()).unwrap(); } let filter = LogFilter { diff --git a/src/eth/storage/permanent/rocks/types/account.rs b/src/eth/storage/permanent/rocks/types/account.rs index 87dbaf41b..a0e33258a 100644 --- a/src/eth/storage/permanent/rocks/types/account.rs +++ b/src/eth/storage/permanent/rocks/types/account.rs @@ -4,6 +4,9 @@ use super::address::AddressRocksdb; use super::bytecode::BytecodeRocksdb; use super::nonce::NonceRocksdb; use super::wei::WeiRocksdb; +use crate::eth::executor::types::state::AccountChanges; +use crate::eth::executor::types::state::Change; +use crate::eth::executor::types::state::Final; use crate::eth::storage::permanent::rocks::SerializeDeserializeWithContext; use crate::eth::types::Account; use crate::eth::types::Address; @@ -26,6 +29,22 @@ impl AccountRocksdb { bytecode: self.bytecode.clone().map_into(), } } + + pub fn update(mut self, other: AccountChanges) -> Self { + if other.balance.is_changed() { + self.balance = other.balance.take_value().into(); + } + + if other.nonce.is_changed() { + self.nonce = other.nonce.take_value().into(); + } + + if other.bytecode.is_changed() { + self.bytecode = other.bytecode.take_value().map_into(); + } + + self + } } impl From for (AddressRocksdb, AccountRocksdb) { diff --git a/src/eth/storage/permanent/rocks/types/block_changes.rs b/src/eth/storage/permanent/rocks/types/block_changes.rs index 85934d0ac..212d9d67b 100644 --- a/src/eth/storage/permanent/rocks/types/block_changes.rs +++ b/src/eth/storage/permanent/rocks/types/block_changes.rs @@ -2,12 +2,16 @@ use std::collections::HashMap; use serde_with::serde_as; +use crate::eth::executor::types::state::AccountChanges; +use crate::eth::executor::types::state::Change; +use crate::eth::executor::types::state::Final; use crate::eth::storage::permanent::rocks::types::AddressRocksdb; use crate::eth::storage::permanent::rocks::types::SlotIndexRocksdb; use crate::eth::storage::permanent::rocks::types::SlotValueRocksdb; use crate::eth::storage::permanent::rocks::types::bytecode::BytecodeRocksdb; use crate::eth::storage::permanent::rocks::types::nonce::NonceRocksdb; use crate::eth::storage::permanent::rocks::types::wei::WeiRocksdb; +use crate::ext::OptionExt; #[derive(Debug, Clone, PartialEq, bincode::Encode, bincode::Decode, serde::Serialize, serde::Deserialize, Default)] #[cfg_attr(test, derive(fake::Dummy))] @@ -46,3 +50,13 @@ impl From<()> for BlockChangesRocksdb { unimplemented!() } } + +impl From<&AccountChanges> for AccountChangesRocksdb { + fn from(value: &AccountChanges) -> Self { + Self { + balance: value.balance.changed_ref().copied().map_into(), + nonce: value.nonce.changed_ref().copied().map_into(), + bytecode: value.bytecode.changed_ref().cloned().map(|opt| opt.map_into()), + } + } +} diff --git a/src/eth/storage/permanent/rocks/types/transaction_mined.rs b/src/eth/storage/permanent/rocks/types/transaction_mined.rs index 733425170..1372584b9 100644 --- a/src/eth/storage/permanent/rocks/types/transaction_mined.rs +++ b/src/eth/storage/permanent/rocks/types/transaction_mined.rs @@ -6,10 +6,9 @@ use super::hash::HashRocksdb; use super::index::IndexRocksdb; use super::log_mined::LogMinedRocksdb; use super::transaction_input::TransactionInputRocksdb; -use crate::eth::executor::Changes; use crate::eth::executor::TransactionExecution; use crate::eth::executor::TransactionExecutionInput; -use crate::eth::executor::TransactionExecutionOutput; +use crate::eth::executor::TransactionExecutionResult; use crate::eth::storage::permanent::rocks::SerializeDeserializeWithContext; use crate::eth::storage::permanent::rocks::types::execution_result::ExecutionResultBuilder; use crate::eth::types::Index; @@ -34,29 +33,29 @@ impl From for TransactionMinedRocksdb { Self { input: TransactionInputRocksdb { tx_type: execution.info.tx_type.map(|inner| inner.as_u64() as u8), - chain_id: execution.evm_input.chain_id.map_into(), + chain_id: execution.input.chain_id.map_into(), hash: execution.info.hash.into(), - nonce: execution.evm_input.nonce.into(), - signer: execution.evm_input.from.into(), - from: execution.evm_input.from.into(), - to: execution.evm_input.to.map_into(), - value: execution.evm_input.value.into(), - input: execution.evm_input.data.clone().into(), - gas_limit: execution.evm_input.gas_limit.into(), - gas_price: execution.evm_input.gas_price.into(), + nonce: execution.input.nonce.into(), + signer: execution.input.from.into(), + from: execution.input.from.into(), + to: execution.input.to.map_into(), + value: execution.input.value.into(), + input: execution.input.data.clone().into(), + gas_limit: execution.input.gas_limit.into(), + gas_price: execution.input.gas_price.into(), v: execution.signature.v.as_u64(), r: execution.signature.r.into_limbs(), s: execution.signature.s.into_limbs(), }, execution: ExecutionRocksdb::new( - execution.evm_input.block_timestamp.into(), - execution.result.result.into(), - execution.result.output.into(), - execution.result.gas_used.into(), - execution.result.deployed_contract_address.map_into(), + execution.input.block_timestamp.into(), + execution.output.result.into(), + execution.output.output.into(), + execution.output.gas_used.into(), + execution.output.deployed_contract_address.map_into(), ), logs: execution - .result + .output .logs .into_iter() .enumerate() @@ -80,12 +79,11 @@ impl TransactionMined { let (result, output) = ExecutionResultBuilder((other.execution.result, other.execution.output)).build(); let input = TransactionInput::from(other.input); - let evm_result = TransactionExecutionOutput { + let evm_result = TransactionExecutionResult { result, output, logs, gas_used: other.execution.gas.into(), - changes: Changes::default(), deployed_contract_address: other.execution.deployed_contract_address.map_into(), }; @@ -93,8 +91,8 @@ impl TransactionMined { let execution = TransactionExecution { info: input.transaction_info, signature: input.signature, - evm_input, - result: evm_result, + input: evm_input, + output: evm_result, }; Self { execution, mined_data } diff --git a/src/eth/storage/resolve_pending.rs b/src/eth/storage/resolve_pending.rs index 3579cd96b..1909558fa 100644 --- a/src/eth/storage/resolve_pending.rs +++ b/src/eth/storage/resolve_pending.rs @@ -1,9 +1,7 @@ use parking_lot::RwLockReadGuard; use crate::eth::storage::ExecutionKind; -use crate::eth::storage::StorageError; use crate::eth::storage::StratusStorage; -use crate::eth::storage::TxCount; use crate::eth::storage::stratus_storage::EntityRead; use crate::eth::types::Account; use crate::eth::types::BlockNumber; @@ -35,11 +33,11 @@ pub enum MinedPointInTime<'a> { } impl<'a> MinedPointInTime<'a> { - fn mined(guard: Option>) -> Self { + fn latest(guard: Option>) -> Self { Self::Latest(Seal(SealPrivate), guard) } - fn mined_past(number: BlockNumber) -> Self { + fn past(number: BlockNumber) -> Self { Self::Past(Seal(SealPrivate), number) } @@ -70,8 +68,6 @@ impl<'a> Drop for MinedPointInTime<'a> { /// Outcome of resolving pending state for a read. #[derive(Debug)] pub(super) enum Resolved<'a, T> { - /// Found in the pending cache. - PendingCache(T), /// Found in temporary storage. Temp(T), /// Nothing pending. @@ -80,18 +76,13 @@ pub(super) enum Resolved<'a, T> { /// Pending-state resolution, generic over the entity being read. pub(super) trait Resolve: EntityRead { - fn resolve(s: &StratusStorage, key: Self::Key, kind: ExecutionKind) -> Result, StorageError> { - if kind.point_in_time() == PointInTime::Pending { - if matches!(kind, ExecutionKind::Transaction) - && let Some(value) = Self::read_pending_cache(s, key) - { - return Ok(Resolved::PendingCache(value)); - } - if let Some(value) = Self::read_temp(s, key, kind)? { - return Ok(Resolved::Temp(value)); - } + fn resolve(s: &StratusStorage, key: Self::Key, kind: ExecutionKind) -> Resolved<'_, Self> { + if kind.point_in_time() == PointInTime::Pending + && let Some(value) = Self::read_temp(s, key) + { + return Resolved::Temp(value); } - Ok(Resolved::Miss(s.resolve_mined_point(kind))) + Resolved::Miss(s.resolve_mined_point(kind)) } } @@ -100,28 +91,22 @@ impl Resolve for Account {} impl Resolve for Slot {} impl StratusStorage { - fn resolve_call_point(&self, block_number: BlockNumber, tx_count: TxCount) -> MinedPointInTime<'_> { + fn resolve_call_point(&self, block_number: BlockNumber) -> MinedPointInTime<'_> { let guard = self.transient_state_lock.read(); let mined = self.read_mined_block_number(); - if (block_number, tx_count) >= (mined, TxCount::Full) { - MinedPointInTime::mined(Some(guard)) + if block_number >= mined { + MinedPointInTime::latest(Some(guard)) } else { - drop(guard); - let target = match tx_count { - TxCount::Partial(_) => block_number.prev().unwrap_or_default(), - TxCount::Full => block_number, - }; - MinedPointInTime::mined_past(target) + MinedPointInTime::past(block_number) } } /// Determines the mined point-in-time for a read. fn resolve_mined_point(&self, kind: ExecutionKind) -> MinedPointInTime<'_> { match kind { - ExecutionKind::RPC(PointInTime::Past(number)) | ExecutionKind::CallPast(number) => MinedPointInTime::mined_past(number), - ExecutionKind::CallPending(block_number, tx_count) => self.resolve_call_point(block_number, tx_count), - ExecutionKind::CallLatest(block_number) => self.resolve_call_point(block_number, TxCount::Full), - ExecutionKind::Transaction | ExecutionKind::RPC(_) => MinedPointInTime::mined(None), + ExecutionKind::RPC(PointInTime::Past(number)) | ExecutionKind::CallPast(number) => MinedPointInTime::past(number), + ExecutionKind::CallLatest(block_number) => self.resolve_call_point(block_number), + ExecutionKind::Transaction | ExecutionKind::RPC(_) => MinedPointInTime::latest(None), } } } @@ -131,65 +116,11 @@ mod tests { use super::super::StratusStorage; use super::Resolve; use crate::eth::storage::ExecutionKind; - use crate::eth::storage::TxCount; use crate::eth::types::Address; use crate::eth::types::BlockNumber; use crate::eth::types::Slot; use crate::eth::types::SlotIndex; - #[test] - fn pending_partial_call_latest_becomes_stale_once_block_is_mined() { - let storage = StratusStorage::new_test().expect("failed to build test storage"); - - let address = Address::ZERO; - let index = SlotIndex::ZERO; - - // Pending call: block_number is the pending block (mined + 1 = 6), pinned to tx 0. - let mined_at_start = 5u64; - let pending_block_number = BlockNumber::from(mined_at_start + 1); - storage.set_mined_block_number(BlockNumber::from(mined_at_start)); - - let kind = ExecutionKind::CallPending(pending_block_number, TxCount::Partial(0)); - - // At call start: block 6 is still pending (mined=5). Latest (block 5) is a safe base. - // resolve_slot should return Miss(Mined(Some(guard))). - let resolved = Slot::resolve(&storage, (address, index), kind).expect("resolve_slot"); - match resolved { - super::Resolved::Miss(mut point) => { - assert!( - matches!(point, super::MinedPointInTime::Latest(_, _)), - "should read latest mined base while block is still pending" - ); - assert!(point.take_guard().is_some(), "guard should be held for valid latest read"); - } - other => panic!("expected Miss, got {other:?}"), - } - - // The pending block is mined mid-call, advancing the mined tip to 6. - storage.set_mined_block_number(pending_block_number); - - // The call is now stale: reading "latest" would observe block 6's aggregate state, - // not the tx-0 base. resolve_slot should downgrade to MinedPast(5) = b.prev() with no guard. - let resolved = Slot::resolve(&storage, (address, index), kind).expect("resolve_slot"); - match resolved { - super::Resolved::Miss(mut point) => { - assert!(!matches!(point, super::MinedPointInTime::Latest(_, _)), "stale call should not read latest"); - match &point { - super::MinedPointInTime::Past(_, number) => { - assert_eq!( - *number, - BlockNumber::from(mined_at_start), - "stale Partial call should downgrade to MinedPast(b.prev())" - ); - } - other => panic!("expected Past, got {other:?}"), - } - assert!(point.take_guard().is_none(), "no guard for historical read"); - } - other => panic!("expected Miss, got {other:?}"), - } - } - #[test] fn mined_full_call_downgrades_to_minedpast_block_not_prev() { let storage = StratusStorage::new_test().expect("failed to build test storage"); @@ -203,7 +134,7 @@ mod tests { let kind = ExecutionKind::CallLatest(call_block); - let resolved = Slot::resolve(&storage, (address, index), kind).expect("resolve_slot"); + let resolved = Slot::resolve(&storage, (address, index), kind); match resolved { super::Resolved::Miss(mut point) => { assert!( @@ -219,7 +150,7 @@ mod tests { storage.set_mined_block_number(BlockNumber::from(6u64)); // Stale: b=5 < mined=6. Full → MinedPast(5), NOT MinedPast(4). - let resolved = Slot::resolve(&storage, (address, index), kind).expect("resolve_slot"); + let resolved = Slot::resolve(&storage, (address, index), kind); match resolved { super::Resolved::Miss(mut point) => { assert!(!matches!(point, super::MinedPointInTime::Latest(_, _)), "stale call should not read latest"); diff --git a/src/eth/storage/stratus_storage.rs b/src/eth/storage/stratus_storage.rs index 9c00c9cab..b2d62ecfb 100644 --- a/src/eth/storage/stratus_storage.rs +++ b/src/eth/storage/stratus_storage.rs @@ -1,8 +1,10 @@ use tracing::Span; -use crate::eth::executor::AccountOriginalsReader; -use crate::eth::executor::Changes; +use crate::eth::executor::State; use crate::eth::executor::TransactionExecution; +use crate::eth::executor::types::state::AccountOriginalsReader; +use crate::eth::executor::types::state::Complete; +use crate::eth::executor::types::state::Final; #[cfg(feature = "dev")] use crate::eth::genesis::GenesisConfig; use crate::eth::rpc::BlockFilter; @@ -12,7 +14,6 @@ use crate::eth::storage::InMemoryTemporaryStorage; use crate::eth::storage::RocksPermanentStorage; use crate::eth::storage::StorageCache; use crate::eth::storage::StorageError; -use crate::eth::storage::TxCount; use crate::eth::storage::permanent::rocks::types::BlockChangesRocksdb; use crate::eth::storage::permanent::rocks::types::BlockRocksdb; use crate::eth::storage::resolve_pending; @@ -88,17 +89,12 @@ enum FoundAt { /// Abstraction over address-keyed ([`Account`]) and slot-keyed ([`Slot`]) reads pub(super) trait EntityRead: Sized + Clone { type Key: Copy; - - /// Reads the pending (current block) value from the cache, if present. - fn read_pending_cache(s: &StratusStorage, key: Self::Key) -> Option; /// Reads the latest (mined tip) value from the cache, if present. fn read_latest_cache(s: &StratusStorage, key: Self::Key) -> Option; /// Reads from temporary (pending) storage. - fn read_temp(s: &StratusStorage, key: Self::Key, kind: ExecutionKind) -> Result, StorageError>; + fn read_temp(s: &StratusStorage, key: Self::Key) -> Option; /// Reads from permanent storage at the resolved mined point. fn read_perm(s: &StratusStorage, key: Self::Key, point: MinedPointInTime<'_>) -> Result; - /// Caches the value as a pending entry, if not already cached. - fn cache_if_missing(s: &StratusStorage, key: Self::Key, value: Self); /// Caches the value as a latest (mined tip) entry, if not already cached. fn cache_latest_if_missing(s: &StratusStorage, key: Self::Key, value: Self); } @@ -106,24 +102,12 @@ pub(super) trait EntityRead: Sized + Clone { impl EntityRead for Account { type Key = Address; - fn read_pending_cache(s: &StratusStorage, address: Address) -> Option { - timed(|| s.cache.get_account(address)).with(|m| { - if m.result.is_some() { - tracing::debug!(storage = %label::CACHE, %address, "account found in cache"); - metrics::inc_storage_read_account(m.elapsed, label::CACHE, PointInTime::Pending, true); - } - }) - } - - fn read_temp(s: &StratusStorage, address: Address, kind: ExecutionKind) -> Result, StorageError> { + fn read_temp(s: &StratusStorage, address: Address) -> Option { tracing::debug!(storage = %label::TEMP, %address, "reading account"); - timed(|| s.temp.read_account(address, kind)).with(|m| { - if m.result.as_ref().is_ok_and(|opt| opt.is_some()) { + timed(|| s.temp.read_account(address)).with(|m| { + if m.result.is_some() { metrics::inc_storage_read_account(m.elapsed, label::TEMP, PointInTime::Pending, true); } - if let Err(ref e) = m.result { - tracing::error!(reason = ?e, "failed to read account from temporary storage"); - } }) } @@ -159,10 +143,6 @@ impl EntityRead for Account { }) } - fn cache_if_missing(s: &StratusStorage, _address: Address, account: Self) { - s.cache.cache_account_if_missing(account); - } - fn cache_latest_if_missing(s: &StratusStorage, address: Address, account: Self) { s.cache.cache_account_latest_if_missing(address, account); } @@ -171,26 +151,13 @@ impl EntityRead for Account { impl EntityRead for Slot { type Key = (Address, SlotIndex); - fn read_pending_cache(s: &StratusStorage, key: (Address, SlotIndex)) -> Option { - let (address, index) = key; - timed(|| s.cache.get_slot(address, index)).with(|m| { - if m.result.is_some() { - tracing::debug!(storage = %label::CACHE, %address, slot = ?m.result, "slot found in cache"); - metrics::inc_storage_read_slot(m.elapsed, label::CACHE, PointInTime::Pending, true); - } - }) - } - - fn read_temp(s: &StratusStorage, key: (Address, SlotIndex), kind: ExecutionKind) -> Result, StorageError> { + fn read_temp(s: &StratusStorage, key: (Address, SlotIndex)) -> Option { let (address, index) = key; tracing::debug!(storage = %label::TEMP, %address, %index, "reading slot"); - timed(|| s.temp.read_slot(address, index, kind)).with(|m| { - if m.result.as_ref().is_ok_and(|opt| opt.is_some()) { + timed(|| s.temp.read_slot(address, index)).with(|m| { + if m.result.is_some() { metrics::inc_storage_read_slot(m.elapsed, label::TEMP, PointInTime::Pending, true); } - if let Err(ref e) = m.result { - tracing::error!(reason = ?e, "failed to read slot from temporary storage"); - } }) } @@ -228,11 +195,6 @@ impl EntityRead for Slot { }) } - fn cache_if_missing(s: &StratusStorage, key: (Address, SlotIndex), slot: Self) { - let (address, _) = key; - s.cache.cache_slot_if_missing(address, slot); - } - fn cache_latest_if_missing(s: &StratusStorage, key: (Address, SlotIndex), slot: Self) { let (address, _) = key; s.cache.cache_slot_latest_if_missing(address, slot); @@ -299,8 +261,6 @@ impl StratusStorage { .expect("Failed to create RocksPermanentStorage for tests"); let cache = CacheConfig { - slot_cache_capacity: 100000, - account_cache_capacity: 20000, account_history_cache_capacity: 20000, slot_history_cache_capacity: 100000, } @@ -327,13 +287,13 @@ impl StratusStorage { #[cfg(feature = "tracing")] let _span = tracing::info_span!("storage::read_block_number_to_resume_import").entered(); - let number = self.read_pending_block_header().0.number; + let number = self.read_pending_block_header().number; tracing::info!(?number, "got block number to resume import"); Ok(number) } - pub fn read_pending_block_header(&self) -> (PendingBlockHeader, TxCount) { + pub fn read_pending_block_header(&self) -> PendingBlockHeader { #[cfg(feature = "tracing")] let _span = tracing::info_span!("storage::read_pending_block_number").entered(); tracing::debug!(storage = %label::TEMP, "reading pending block number"); @@ -396,8 +356,7 @@ impl StratusStorage { /// Generic read algorithm shared by [`read_account`] and [`read_slot`]. fn read(&self, key: E::Key, kind: ExecutionKind) -> Result { let (value, found_at) = 'query: { - match E::resolve(self, key, kind)? { - resolve_pending::Resolved::PendingCache(value) => break 'query (value, FoundAt::Cache), + match E::resolve(self, key, kind) { resolve_pending::Resolved::Temp(value) => break 'query (value, FoundAt::Temp), resolve_pending::Resolved::Miss(mined_point) => { let found_at = match &mined_point { @@ -407,6 +366,7 @@ impl StratusStorage { if let Some(value) = E::read_latest_cache(self, key) { break 'query (value, FoundAt::Cache); } + // If it wasnt found in the cache and we still have the guard the value can only be read in perm latest FoundAt::PermLatest } MinedPointInTime::Past(_, _) => FoundAt::PermHistorical, @@ -417,18 +377,11 @@ impl StratusStorage { }; // Cache non-historical reads according to the point-in-time and where the value came from. - match (kind.point_in_time(), found_at) { - // A pending read that hit perm (i.e. not in any cache/temp) is already mined, so cache in both. - (PointInTime::Pending, FoundAt::PermLatest) => { - E::cache_if_missing(self, key, value.clone()); - E::cache_latest_if_missing(self, key, value.clone()); - } - // A pending read that hit temp was not found in the pending cache, so populate it. - (PointInTime::Pending, FoundAt::Temp) => { - E::cache_if_missing(self, key, value.clone()); - } - // A mined read that hit perm is the latest state, so populate the latest cache. - (PointInTime::Latest, FoundAt::PermLatest) => { + match (kind, found_at) { + (ExecutionKind::Transaction, _) => (), + // A pending read that hit perm (i.e. not in any cache/temp) is already mined, so cache latest. + // OR A mined read that hit perm is the latest state, so populate the latest cache. + (_, FoundAt::PermLatest) => { E::cache_latest_if_missing(self, key, value.clone()); } // Cache / Historical / (Mined, Temp): nothing to cache. @@ -453,34 +406,30 @@ impl StratusStorage { // Blocks // ------------------------------------------------------------------------- - pub fn save_execution(&self, tx: TransactionExecution) -> Result<(), StorageError> { - let changes = tx.result.changes.clone(); - + pub fn save_execution(&self, tx: TransactionExecution, state: State) -> Result<(), StorageError> { #[cfg(feature = "tracing")] let _span = tracing::info_span!("storage::save_execution", tx_hash = %tx.info.hash).entered(); - tracing::debug!(storage = %label::TEMP, tx_hash = %tx.info.hash, changes = ?tx.result.changes, "saving execution"); + tracing::debug!(storage = %label::TEMP, tx_hash = %tx.info.hash, changes = ?state, "saving execution"); // Log warning if a failed transaction has slot changes - if !tx.result.result.is_success() { - let total_slot_changes: usize = changes.slots.len(); + if !tx.output.result.is_success() { + let total_slot_changes: usize = state.slots.len(); if total_slot_changes > 0 { tracing::warn!(?tx, "Failed transaction contains {} slot change(s)", total_slot_changes); } } - timed(|| self.temp.save_pending_execution(tx)) - .with(|m| { - metrics::inc_storage_save_execution(m.elapsed, label::TEMP, m.result.is_ok()); - match &m.result { - Err(StorageError::EvmInputMismatch { .. }) => { - tracing::warn!("failed to save execution due to mismatch, will retry"); - } - Err(e) => tracing::error!(reason = ?e, "failed to save execution"), - _ => (), + timed(|| self.temp.save_pending_execution(tx, state)).with(|m| { + metrics::inc_storage_save_execution(m.elapsed, label::TEMP, m.result.is_ok()); + match &m.result { + Err(StorageError::EvmInputMismatch { .. }) => { + tracing::warn!("failed to save execution due to mismatch, will retry"); } - }) - .inspect(|_| self.cache.cache_account_and_slots_from_changes(changes)) + Err(e) => tracing::error!(reason = ?e, "failed to save execution"), + _ => (), + } + }) } /// Retrieves pending transactions being mined. @@ -488,26 +437,21 @@ impl StratusStorage { self.temp.read_pending_executions() } - pub fn finish_pending_block(&self) -> Result<(PendingBlock, Changes), StorageError> { + pub fn finish_pending_block(&self) -> (PendingBlock, State) { #[cfg(feature = "tracing")] let _span = tracing::info_span!("storage::finish_pending_block", block_number = tracing::field::Empty).entered(); tracing::debug!(storage = %label::TEMP, "finishing pending block"); let result = timed(|| self.temp.finish_pending_block()).with(|m| { - metrics::inc_storage_finish_pending_block(m.elapsed, label::TEMP, m.result.is_ok()); - if let Err(ref e) = m.result { - tracing::error!(reason = ?e, "failed to finish pending block"); - } + metrics::inc_storage_finish_pending_block(m.elapsed); }); - if let Ok((ref block, _)) = result { - Span::with(|s| s.rec_str("block_number", &block.header.number)); - } + Span::with(|s| s.rec_str("block_number", &result.0.header.number)); result } - pub fn save_genesis_block(&self, block: Block, accounts: Vec, changes: Changes) -> Result<(), StorageError> { + pub fn save_genesis_block(&self, block: Block, accounts: Vec, changes: State) -> Result<(), StorageError> { let block_number = block.number(); #[cfg(feature = "tracing")] @@ -523,7 +467,7 @@ impl StratusStorage { }) } - pub fn save_block(&self, block: Block, changes: Changes) -> Result<(), StorageError> { + pub fn save_block(&self, block: Block, changes: State) -> Result<(), StorageError> { let block_number = block.number(); #[cfg(feature = "tracing")] @@ -542,11 +486,11 @@ impl StratusStorage { // check pending number let pending_header = self.read_pending_block_header(); - if block_number >= pending_header.0.number { - tracing::error!(%block_number, pending_number = %pending_header.0.number, "failed to save block because mismatch with pending block number"); + if block_number >= pending_header.number { + tracing::error!(%block_number, pending_number = %pending_header.number, "failed to save block because mismatch with pending block number"); return Err(StorageError::PendingNumberConflict { new: block_number, - pending: pending_header.0.number, + pending: pending_header.number, }); } @@ -561,8 +505,8 @@ impl StratusStorage { timed(|| { let guard = self.transient_state_lock.write(); - self.perm.save_block(block, changes.clone())?; - self.cache.cache_account_and_slots_latest_from_changes(changes); + self.cache.cache_account_and_slots_latest_from_changes(&changes); + self.perm.save_block(block, changes.finalize())?; drop(guard); Ok(()) }) @@ -808,7 +752,7 @@ impl StratusStorage { } }; // Save the genesis block - self.save_block(genesis_block, Changes::default())?; + self.save_block(genesis_block, State::default())?; // accounts self.save_accounts(genesis_accounts)?; @@ -849,10 +793,11 @@ impl StratusStorage { #[cfg(test)] mod tests { use super::*; - use crate::eth::executor::AccountChanges; use crate::eth::executor::ExecutionResult; use crate::eth::executor::TransactionExecutionInput; - use crate::eth::executor::TransactionExecutionOutput; + use crate::eth::executor::TransactionExecutionResult; + use crate::eth::executor::types::state::AccountChanges; + use crate::eth::executor::types::state::CompleteValue; use crate::eth::types::Signature; use crate::eth::types::SlotValue; use crate::eth::types::TransactionInfo; @@ -860,20 +805,19 @@ mod tests { use crate::eth::types::Wei; /// Mines a block applying `changes` - fn mine_block(storage: &StratusStorage, changes: Changes) -> BlockNumber { - let (header, _) = storage.read_pending_block_header(); + fn mine_block(storage: &StratusStorage, changes: State) -> BlockNumber { + let header = storage.read_pending_block_header(); let evm_input = TransactionExecutionInput::from_eth_transaction(&TransactionInput::default(), header.number, *header.timestamp); - let result = TransactionExecutionOutput { + let result = TransactionExecutionResult { result: ExecutionResult::Success, - changes, ..Default::default() }; let tx = TransactionExecution::new(TransactionInfo::default(), Signature::default(), evm_input, result); - storage.save_execution(tx).expect("save execution"); + storage.save_execution(tx, changes).expect("save execution"); - let (block, block_changes) = storage.finish_pending_block().expect("finish pending block"); + let (block, block_changes) = storage.finish_pending_block(); storage.save_block(block.into(), block_changes).expect("save block"); storage.read_mined_block_number() @@ -889,13 +833,17 @@ mod tests { let index = SlotIndex::ZERO; // Mine a block setting slot S = 100. The eth_call captures this block. - let mut changes1 = Changes::default(); - changes1.slots.insert((address, index), SlotValue::from([100u64, 0, 0, 0])); + let mut changes1 = State::default(); + changes1 + .slots + .insert((address, index), CompleteValue::Changed(SlotValue::from([100u64, 0, 0, 0]))); let call_block = mine_block(&storage, changes1); // A new block is mined while the call is in flight, changing the slot to 200. - let mut changes2 = Changes::default(); - changes2.slots.insert((address, index), SlotValue::from([200u64, 0, 0, 0])); + let mut changes2 = State::default(); + changes2 + .slots + .insert((address, index), CompleteValue::Changed(SlotValue::from([200u64, 0, 0, 0]))); let latest = mine_block(&storage, changes2); assert_ne!(call_block, latest); @@ -913,14 +861,14 @@ mod tests { let address = Address::new([0xBB; 20]); // Mine a block setting the account balance to 100. The eth_call captures this block. - let mut changes1 = Changes::default(); + let mut changes1 = State::default(); changes1 .accounts .insert(address, AccountChanges::from_changed(Account::new_with_balance(address, Wei::from(100u64)))); let call_block = mine_block(&storage, changes1); // A new block is mined while the call is in flight, changing the balance to 200. - let mut changes2 = Changes::default(); + let mut changes2 = State::default(); changes2 .accounts .insert(address, AccountChanges::from_changed(Account::new_with_balance(address, Wei::from(200u64)))); diff --git a/src/eth/storage/temporary/inmemory/call.rs b/src/eth/storage/temporary/inmemory/call.rs deleted file mode 100644 index da4dcf454..000000000 --- a/src/eth/storage/temporary/inmemory/call.rs +++ /dev/null @@ -1,127 +0,0 @@ -//! In-memory call temporary storage implementation. - -use std::collections::HashMap; - -use dashmap::DashMap; - -use crate::eth::executor::TransactionExecution; -use crate::eth::storage::temporary::inmemory::TxCount; -use crate::eth::types::Account; -use crate::eth::types::Address; -use crate::eth::types::BlockNumber; -use crate::eth::types::Slot; -use crate::eth::types::SlotIndex; -use crate::eth::types::SlotValue; - -#[derive(Debug)] -pub struct InMemoryCallTemporaryStorage { - /// Storage for call temporary data indexed by (block_number, tx_count). - /// tx_count is None for block-level data, Some(count) for transaction-level data. - storage: DashMap, -} - -impl InMemoryCallTemporaryStorage { - /// Creates a new instance of InmemoryCallTemporaryStorage. - pub fn new() -> Self { - Self { - storage: DashMap::with_capacity(10), - } - } - - /// Reads the latest account data for the given address at the specified block and transaction. - /// - /// Returns the account data if found, otherwise None. - pub fn read_account(&self, block: BlockNumber, tx: TxCount, address: Address) -> Option { - if let Some(block_state) = self.storage.get(&block) - && let Some(accounts) = block_state.accounts.get(&address) - { - let vec_index = match accounts.binary_search_by_key(&&tx, |(_, tx_count)| tx_count) { - Ok(index) => index, - Err(index) => index - 1, - }; - return accounts.get(vec_index).map(|(acc, _)| acc.clone()); - } - None - } - - /// Reads the latest slot value for the given address and slot index at the specified block and transaction. - /// - /// Returns the slot value if found, otherwise None. - pub fn read_slot(&self, block: BlockNumber, tx: TxCount, address: Address, slot: SlotIndex) -> Option { - if let Some(block_state) = self.storage.get(&block) - && let Some(slot_values) = block_state.slots.get(&(address, slot)) - { - let vec_index = match slot_values.binary_search_by_key(&&tx, |(_, tx_count)| tx_count) { - Ok(index) => index, - Err(index) => index - 1, - }; - - return slot_values.get(vec_index).map(|(value, _)| Slot { index: slot, value: *value }); - } - None - } - - /// Updates the storage by appending new entries from the given TransactionExecution. - /// - /// This function takes a transaction execution and appends all account and slot changes - /// to the corresponding BlockNumber in the call storage with the given transaction count. - pub fn update_state_with_transaction(&self, tx: &TransactionExecution) { - let block_number = tx.evm_input.block_number; - - // Get or create the block state - let mut block_state = self.storage.entry(block_number).or_default(); - block_state.current_tx_count += 1; - let current_tx_count = block_state.current_tx_count; - - // Process each account change from the transaction execution - for (address, change) in &tx.result.changes.accounts { - if change.is_modified() { - // Build the account from the changes - let account = (*address, change.clone()).into(); - block_state.accounts.entry(*address).or_default().push((account, current_tx_count)); - } - } - - // Add slot changes - for ((address, slot_index), slot_value) in &tx.result.changes.slots { - block_state - .slots - .entry((*address, *slot_index)) - .or_default() - .push((*slot_value, current_tx_count)); - } - } - - /// Retains only the 10 most recent blocks and removes all older entries. - pub fn retain_recent_blocks(&self) { - let mut block_numbers: Vec = self.storage.iter().map(|r| *r.key()).collect(); - block_numbers.sort_by(|a, b| b.cmp(a)); // Sort in descending order - - // If we have more than 10 blocks, determine the cutoff point - if block_numbers.len() > 9 { - let cutoff = block_numbers[9]; - // Remove all blocks older than or equal to the cutoff - self.storage.retain(|block_number, _| *block_number > cutoff); - } - } - - /// Clears all data from the call temporary storage. - /// - /// This function removes all stored block state data, effectively resetting the storage. - pub fn reset(&self) { - self.storage.clear(); - } -} - -impl Default for InMemoryCallTemporaryStorage { - fn default() -> Self { - Self::new() - } -} - -#[derive(Debug, Default)] -pub struct InMemoryCallTemporaryStorageState { - pub current_tx_count: TxCount, - pub accounts: HashMap, hash_hasher::HashBuildHasher>, - pub slots: HashMap<(Address, SlotIndex), Vec<(SlotValue, TxCount)>, hash_hasher::HashBuildHasher>, -} diff --git a/src/eth/storage/temporary/inmemory/mod.rs b/src/eth/storage/temporary/inmemory/mod.rs index 016f8ca7c..d3530a04e 100644 --- a/src/eth/storage/temporary/inmemory/mod.rs +++ b/src/eth/storage/temporary/inmemory/mod.rs @@ -1,11 +1,9 @@ //! In-memory storage implementations. -use crate::eth::executor::Changes; +use crate::eth::executor::State; use crate::eth::executor::TransactionExecution; -use crate::eth::storage::ExecutionKind; +use crate::eth::executor::types::state::Complete; use crate::eth::storage::StorageError; -use crate::eth::storage::TxCount; -use crate::eth::storage::temporary::inmemory::call::InMemoryCallTemporaryStorage; use crate::eth::storage::temporary::inmemory::transaction::InmemoryTransactionTemporaryStorage; use crate::eth::types::Account; use crate::eth::types::Address; @@ -23,24 +21,21 @@ use crate::eth::types::UnixTime; #[cfg(feature = "dev")] use crate::eth::types::Wei; -mod call; mod transaction; #[derive(Debug)] pub struct InMemoryTemporaryStorage { pub transaction_storage: InmemoryTransactionTemporaryStorage, - pub call_storage: InMemoryCallTemporaryStorage, } impl InMemoryTemporaryStorage { pub fn new(block_number: BlockNumber) -> Self { Self { transaction_storage: InmemoryTransactionTemporaryStorage::new(block_number), - call_storage: InMemoryCallTemporaryStorage::new(), } } - pub fn read_pending_block_header(&self) -> (PendingBlockHeader, TxCount) { + pub fn read_pending_block_header(&self) -> PendingBlockHeader { self.transaction_storage.read_pending_block_header() } @@ -53,17 +48,15 @@ impl InMemoryTemporaryStorage { self.transaction_storage.set_pending_header(number, timestamp); } - pub fn save_pending_execution(&self, tx: TransactionExecution) -> Result<(), StorageError> { - self.call_storage.update_state_with_transaction(&tx); - self.transaction_storage.save_pending_execution(tx) + pub fn save_pending_execution(&self, tx: TransactionExecution, state: State) -> Result<(), StorageError> { + self.transaction_storage.save_pending_execution(tx, state) } pub fn read_pending_executions(&self) -> Vec { self.transaction_storage.read_pending_executions() } - pub fn finish_pending_block(&self) -> anyhow::Result<(PendingBlock, Changes), StorageError> { - self.call_storage.retain_recent_blocks(); + pub fn finish_pending_block(&self) -> (PendingBlock, State) { self.transaction_storage.finish_pending_block() } @@ -71,22 +64,12 @@ impl InMemoryTemporaryStorage { self.transaction_storage.read_pending_execution(hash) } - pub fn read_account(&self, address: Address, kind: ExecutionKind) -> anyhow::Result, StorageError> { - match kind { - ExecutionKind::CallPending(block_number, tx_count) => Ok(self.call_storage.read_account(block_number, tx_count, address)), - ExecutionKind::CallLatest(block_number) => Ok(self.call_storage.read_account(block_number, TxCount::Full, address)), - ExecutionKind::CallPast(block_number) => Ok(self.call_storage.read_account(block_number, TxCount::Full, address)), - _ => self.transaction_storage.read_account(address), - } + pub fn read_account(&self, address: Address) -> Option { + self.transaction_storage.read_account(address) } - pub fn read_slot(&self, address: Address, index: SlotIndex, kind: ExecutionKind) -> anyhow::Result, StorageError> { - match kind { - ExecutionKind::CallPending(block_number, tx_count) => Ok(self.call_storage.read_slot(block_number, tx_count, address, index)), - ExecutionKind::CallLatest(block_number) => Ok(self.call_storage.read_slot(block_number, TxCount::Full, address, index)), - ExecutionKind::CallPast(block_number) => Ok(self.call_storage.read_slot(block_number, TxCount::Full, address, index)), - _ => self.transaction_storage.read_slot(address, index), - } + pub fn read_slot(&self, address: Address, index: SlotIndex) -> Option { + self.transaction_storage.read_slot(address, index) } #[cfg(feature = "dev")] @@ -110,7 +93,6 @@ impl InMemoryTemporaryStorage { } pub fn reset(&self) -> anyhow::Result<(), StorageError> { - self.call_storage.reset(); self.transaction_storage.reset() } } @@ -125,19 +107,19 @@ pub struct InMemoryTemporaryStorageState { pub block: PendingBlock, /// Last state of accounts and slots. Can be recreated from the executions inside the pending block. - pub block_changes: Changes, + pub state: State, } impl InMemoryTemporaryStorageState { pub fn new(block_number: BlockNumber) -> Self { Self { block: PendingBlock::new_at_now(block_number), - block_changes: Changes::default(), + state: State::default(), } } pub fn reset(&mut self) { self.block = PendingBlock::new_at_now(1.into()); - self.block_changes = Changes::default(); + self.state = State::default(); } } diff --git a/src/eth/storage/temporary/inmemory/transaction.rs b/src/eth/storage/temporary/inmemory/transaction.rs index 7529717ab..3ffb07115 100644 --- a/src/eth/storage/temporary/inmemory/transaction.rs +++ b/src/eth/storage/temporary/inmemory/transaction.rs @@ -5,11 +5,13 @@ use parking_lot::RwLockUpgradableReadGuard; #[cfg(not(feature = "dev"))] use parking_lot::RwLockWriteGuard; -use crate::eth::executor::Changes; +use crate::eth::executor::State; use crate::eth::executor::TransactionExecution; use crate::eth::executor::TransactionExecutionInput; +use crate::eth::executor::types::state::Complete; +#[cfg(feature = "dev")] +use crate::eth::executor::types::state::CompleteValue; use crate::eth::storage::StorageError; -use crate::eth::storage::TxCount; use crate::eth::storage::temporary::inmemory::InMemoryTemporaryStorageState; use crate::eth::types::Account; use crate::eth::types::Address; @@ -41,7 +43,7 @@ impl InmemoryTransactionTemporaryStorage { Self { pending_block: RwLock::new(InMemoryTemporaryStorageState { block: PendingBlock::new_at_now(block_number), - block_changes: Changes::default(), + state: State::default(), }), latest_block: RwLock::new(None), } @@ -57,10 +59,9 @@ impl InmemoryTransactionTemporaryStorage { // Block number // ------------------------------------------------------------------------- - // Uneeded clone here, return Cow - pub fn read_pending_block_header(&self) -> (PendingBlockHeader, TxCount) { + pub fn read_pending_block_header(&self) -> PendingBlockHeader { let pending_block = self.pending_block.read(); - (pending_block.block.header.clone(), (pending_block.block.transactions.len() as u64).into()) + pending_block.block.header } #[cfg(feature = "dev")] @@ -73,11 +74,11 @@ impl InmemoryTransactionTemporaryStorage { // Block and executions // ------------------------------------------------------------------------- - pub fn save_pending_execution(&self, tx: TransactionExecution) -> Result<(), StorageError> { + pub fn save_pending_execution(&self, tx: TransactionExecution, state: State) -> Result<(), StorageError> { // check conflicts let pending_block = self.pending_block.upgradable_read(); - if tx.evm_input != &pending_block.block.header { - let actual_input = tx.evm_input.clone(); + if tx.input != &pending_block.block.header { + let actual_input = tx.input.clone(); let tx_input: TransactionInput = tx.into(); let expected_input = TransactionExecutionInput::from_eth_transaction(&tx_input, pending_block.block.header.number, *pending_block.block.header.timestamp); @@ -89,7 +90,7 @@ impl InmemoryTransactionTemporaryStorage { let mut pending_block = RwLockUpgradableReadGuard::::upgrade(pending_block); - pending_block.block_changes.merge(tx.result.changes.clone()); // TODO: This clone can be removed by reworking the primitives + pending_block.state.merge(state); // save execution pending_block.block.push_transaction(tx); @@ -106,20 +107,19 @@ impl InmemoryTransactionTemporaryStorage { (*pending_block).clone() } - pub fn finish_pending_block(&self) -> anyhow::Result<(PendingBlock, Changes), StorageError> { + pub fn finish_pending_block(&self) -> (PendingBlock, State) { let pending_block = self.pending_block.upgradable_read(); - let changes = pending_block.block_changes.clone(); // This has to happen BEFORE creating the new state, because UnixTimeNow::default() may change the offset. #[cfg(feature = "dev")] - let finished_block = { + let (finished_block, state) = { let mut finished_block = pending_block.block.clone(); // Update block timestamp only if evm_setNextBlockTimestamp was called, // otherwise keep the original timestamp from pending block creation if UnixTime::evm_set_next_block_timestamp_was_called() { finished_block.header.timestamp = UnixTimeNow::default(); } - finished_block + (finished_block, pending_block.state.clone()) }; let next_state = InMemoryTemporaryStorageState::new(pending_block.block.header.number.next_block_number()); @@ -132,14 +132,15 @@ impl InmemoryTransactionTemporaryStorage { drop(pending_block); #[cfg(not(feature = "dev"))] - let finished_block = { + let (finished_block, state) = { let latest = RwLockWriteGuard::>::downgrade(latest); #[allow(clippy::expect_used)] - latest.as_ref().expect("latest should be Some after finishing the pending block").block.clone() + let latest_state = latest.as_ref().expect("latest should be Some after finishing the pending block"); + (latest_state.block.clone(), latest_state.state.clone()) }; - Ok((finished_block, changes)) + (finished_block, state) } pub fn read_pending_execution(&self, hash: Hash) -> anyhow::Result, StorageError> { @@ -154,27 +155,27 @@ impl InmemoryTransactionTemporaryStorage { // Accounts and Slots // ------------------------------------------------------------------------- - pub fn read_account(&self, address: Address) -> anyhow::Result, StorageError> { - Ok(match self.pending_block.read().block_changes.accounts.get(&address) { + pub fn read_account(&self, address: Address) -> Option { + match self.pending_block.read().state.accounts.get(&address) { Some(pending_account) => Some(pending_account.clone().to_account(address)), None => self .latest_block .read() .as_ref() - .and_then(|latest| latest.block_changes.accounts.get(&address)) + .and_then(|latest| latest.state.accounts.get(&address)) .map(|account| account.clone().to_account(address)), - }) + } } - pub fn read_slot(&self, address: Address, index: SlotIndex) -> anyhow::Result, StorageError> { - Ok(match self.pending_block.read().block_changes.slots.get(&(address, index)) { - Some(pending_value) => Some(Slot::new(index, *pending_value)), + pub fn read_slot(&self, address: Address, index: SlotIndex) -> Option { + match self.pending_block.read().state.slots.get(&(address, index)) { + Some(pending_value) => Some(Slot::new(index, *pending_value.value())), None => self .latest_block .read() .as_ref() - .and_then(|latest| latest.block_changes.slots.get(&(address, index)).map(|value| Slot::new(index, *value))), - }) + .and_then(|latest| latest.state.slots.get(&(address, index)).map(|value| Slot::new(index, *value.value()))), + } } // ------------------------------------------------------------------------- @@ -184,7 +185,10 @@ impl InmemoryTransactionTemporaryStorage { #[cfg(feature = "dev")] pub fn save_slot(&self, address: Address, slot: Slot) -> anyhow::Result<(), StorageError> { let mut pending_block = self.pending_block.write(); - pending_block.block_changes.slots.insert((address, slot.index), slot.value); + pending_block + .block_changes + .slots + .insert((address, slot.index), CompleteValue::Changed(slot.value)); Ok(()) } diff --git a/src/eth/types/block/block.rs b/src/eth/types/block/block.rs index c61017fe6..e3abb0010 100644 --- a/src/eth/types/block/block.rs +++ b/src/eth/types/block/block.rs @@ -96,7 +96,7 @@ impl Block { self.header.hash = external_block.hash(); assert!(*self.header.timestamp == external_block.header.timestamp); for transaction in self.transactions.iter_mut() { - assert!(transaction.evm_input.block_timestamp == self.header.timestamp); + assert!(transaction.input.block_timestamp == self.header.timestamp); transaction.mined_data.block_hash = external_block.hash(); } } @@ -111,9 +111,9 @@ impl From for Block { let mut log_index = Index::ZERO; for (tx_idx, execution) in txs.into_iter().enumerate() { - let log_count = execution.result.logs.len() as u64; + let log_count = execution.output.logs.len() as u64; let transaction_mined = TransactionMined::from_execution(execution, block.hash(), (tx_idx as u64).into(), log_index); - block.header.gas_used += transaction_mined.execution.result.gas_used; + block.header.gas_used += transaction_mined.execution.output.gas_used; block.transactions.push(transaction_mined); log_index += Index(log_count); } diff --git a/src/eth/types/block/pending_block_header.rs b/src/eth/types/block/pending_block_header.rs index 9e8ddb605..94bcd03b9 100644 --- a/src/eth/types/block/pending_block_header.rs +++ b/src/eth/types/block/pending_block_header.rs @@ -4,7 +4,7 @@ use crate::eth::types::BlockNumber; use crate::eth::types::UnixTimeNow; /// Header of the pending block being mined. -#[derive(DebugAsJson, Clone, Default, serde::Serialize)] +#[derive(DebugAsJson, Clone, Copy, Default, serde::Serialize)] pub struct PendingBlockHeader { pub number: BlockNumber, pub timestamp: UnixTimeNow, diff --git a/src/eth/types/execution_kind.rs b/src/eth/types/execution_kind.rs index f07be893a..fed07edbe 100644 --- a/src/eth/types/execution_kind.rs +++ b/src/eth/types/execution_kind.rs @@ -1,11 +1,9 @@ use crate::eth::types::BlockNumber; -use crate::eth::types::Index; use crate::eth::types::PointInTime; #[derive(Clone, Copy, serde::Serialize, PartialEq, Default, Eq)] #[cfg_attr(test, derive(fake::Dummy))] pub enum ExecutionKind { - CallPending(BlockNumber, TxCount), CallLatest(BlockNumber), CallPast(BlockNumber), #[default] @@ -19,61 +17,6 @@ impl ExecutionKind { } } -#[derive(Clone, Copy, PartialEq, Debug, serde::Serialize, Eq)] -#[cfg_attr(test, derive(fake::Dummy))] -pub enum TxCount { - Full, - Partial(u64), -} - -impl TryFrom for Index { - type Error = anyhow::Error; - fn try_from(value: TxCount) -> Result { - match value { - TxCount::Partial(idx) => Ok(idx.into()), - TxCount::Full => anyhow::bail!("full transactions has unknown tx index"), - } - } -} - -impl From for TxCount { - fn from(value: u64) -> Self { - TxCount::Partial(value) - } -} - -impl Default for TxCount { - fn default() -> Self { - TxCount::Partial(0) - } -} - -impl std::ops::AddAssign for TxCount { - fn add_assign(&mut self, rhs: u64) { - match self { - TxCount::Full => {} // If it's Full, keep it Full - TxCount::Partial(count) => *count += rhs, // If it's Partial, increment the counter - } - } -} - -impl Ord for TxCount { - fn cmp(&self, other: &Self) -> std::cmp::Ordering { - match (self, other) { - (TxCount::Full, TxCount::Full) => std::cmp::Ordering::Equal, - (TxCount::Full, TxCount::Partial(_)) => std::cmp::Ordering::Greater, - (TxCount::Partial(_), TxCount::Full) => std::cmp::Ordering::Less, - (TxCount::Partial(a), TxCount::Partial(b)) => a.cmp(b), - } - } -} - -impl PartialOrd for TxCount { - fn partial_cmp(&self, other: &Self) -> Option { - Some(self.cmp(other)) - } -} - impl From<&ExecutionKind> for PointInTime { fn from(value: &ExecutionKind) -> Self { match value { @@ -81,7 +24,6 @@ impl From<&ExecutionKind> for PointInTime { ExecutionKind::Transaction => PointInTime::Pending, ExecutionKind::CallPast(number) => PointInTime::Past(*number), ExecutionKind::CallLatest(_) => PointInTime::Latest, - ExecutionKind::CallPending(_, _) => PointInTime::Pending, } } } diff --git a/src/eth/types/mod.rs b/src/eth/types/mod.rs index 8b55d1767..aec40093c 100644 --- a/src/eth/types/mod.rs +++ b/src/eth/types/mod.rs @@ -14,7 +14,6 @@ pub use error::StateError; pub use error::StratusError; pub use error::UnexpectedError; pub use execution_kind::ExecutionKind; -pub use execution_kind::TxCount; pub use external::ExternalBlock; pub use external::ExternalBlockWithReceipts; pub use external::ExternalReceipt; diff --git a/src/eth/types/primitives/unix_time_now.rs b/src/eth/types/primitives/unix_time_now.rs index 888e29085..68919223a 100644 --- a/src/eth/types/primitives/unix_time_now.rs +++ b/src/eth/types/primitives/unix_time_now.rs @@ -3,7 +3,7 @@ use display_json::DebugAsJson; use crate::eth::types::UnixTime; /// [`UnixTime`] that automatically sets the current time when created. -#[derive(DebugAsJson, Clone, PartialEq, Eq, derive_more::Deref, serde::Serialize)] +#[derive(DebugAsJson, Clone, Copy, PartialEq, Eq, derive_more::Deref, serde::Serialize)] #[cfg_attr(test, derive(serde::Deserialize, fake::Dummy))] pub struct UnixTimeNow(#[deref] UnixTime); diff --git a/src/eth/types/transaction/transaction_mined.rs b/src/eth/types/transaction/transaction_mined.rs index 79903d482..eda0eeab8 100644 --- a/src/eth/types/transaction/transaction_mined.rs +++ b/src/eth/types/transaction/transaction_mined.rs @@ -38,7 +38,7 @@ impl TransactionMined { data: AlloyLogData::new_unchecked(log.topics_non_empty().into_iter().map(Into::into).collect(), log.data.clone().into()), }, block_hash: Some(self.mined_data.block_hash.into()), - block_number: Some(self.evm_input.block_number.as_u64()), + block_number: Some(self.input.block_number.as_u64()), block_timestamp: None, transaction_hash: Some(self.info.hash.into()), transaction_index: Some(*self.mined_data.index), @@ -62,10 +62,10 @@ impl TransactionMined { impl From for AlloyTransaction { fn from(value: TransactionMined) -> Self { - let gas_price = value.execution.evm_input.gas_price; + let gas_price = value.execution.input.gas_price; let block_hash = value.mined_data.block_hash; - let block_number = value.execution.evm_input.block_number; - let block_timestamp = value.execution.evm_input.block_timestamp; + let block_number = value.execution.input.block_number; + let block_timestamp = value.execution.input.block_timestamp; let transaction_index = value.mined_data.index; let tx_input: TransactionInput = value.into(); diff --git a/src/eth/types/transaction/transaction_stage.rs b/src/eth/types/transaction/transaction_stage.rs index 1d9cf9636..5d46650ec 100644 --- a/src/eth/types/transaction/transaction_stage.rs +++ b/src/eth/types/transaction/transaction_stage.rs @@ -2,7 +2,7 @@ use crate::alias::AlloyReceipt; use crate::alias::AlloyTransaction; use crate::alias::JsonValue; use crate::eth::executor::TransactionExecution; -use crate::eth::executor::TransactionExecutionOutput; +use crate::eth::executor::TransactionExecutionResult; use crate::eth::types::MinedData; use crate::eth::types::TransactionMined; use crate::ext::to_json_value; @@ -24,10 +24,10 @@ impl TransactionStage { to_json_value(AlloyTransaction::from(self)) } - pub fn to_result(self) -> TransactionExecutionOutput { + pub fn to_result(self) -> TransactionExecutionResult { match self { - TransactionStage::Mined(tx) => tx.execution.result, - TransactionStage::Pending(tx) => tx.result, + TransactionStage::Mined(tx) => tx.execution.output, + TransactionStage::Pending(tx) => tx.output, } } } diff --git a/src/infra/metrics/metrics_definitions.rs b/src/infra/metrics/metrics_definitions.rs index 160777792..77a912934 100644 --- a/src/infra/metrics/metrics_definitions.rs +++ b/src/infra/metrics/metrics_definitions.rs @@ -72,7 +72,7 @@ metrics! { histogram_duration storage_set_pending_external_block{storage, success}, "Time executing storage finish_pending_block operation." - histogram_duration storage_finish_pending_block{storage, success}, + histogram_duration storage_finish_pending_block{}, "Time executing storage save_block operation." histogram_duration storage_save_block{storage, tens_of_millions_gas_used, success}, diff --git a/src/ledger/events.rs b/src/ledger/events.rs index 1f5c22b55..3d5d2909a 100644 --- a/src/ledger/events.rs +++ b/src/ledger/events.rs @@ -222,7 +222,7 @@ pub fn transaction_to_events(block_timestamp: UnixTime, tx: Cow::from(block_timestamp)); // assert transfers