fix: adapt to updated Reth APIs

Co-authored-by: Derek Cofausper <256792747+decofe@users.noreply.github.com>

Author: legion2002

crates/node/src/rpc.rs

@@ -11,7 +11,7 @@ use std::{
     time::Duration,
 };
 
-use alloy_consensus::BlockHeader;
+use alloy_consensus::{BlockHeader, transaction::TxHashRef};
 use alloy_eips::eip2935::{HISTORY_SERVE_WINDOW, HISTORY_STORAGE_ADDRESS};
 use alloy_network::{ReceiptResponse, TransactionBuilder, TransactionResponse};
 use alloy_primitives::{Address, B256, Bloom, Bytes, U64, U256, keccak256};
@@ -37,7 +37,7 @@ use reth_rpc_eth_api::{
 };
 use reth_rpc_eth_types::{EthApiError, logs_utils};
 use reth_storage_api::{BlockNumReader, StateProviderFactory};
-use reth_trie_common::{ExecutionWitnessMode, HashedStorage};
+use reth_trie_common::{ExecutionWitnessMode, HashedPostState};
 use tempo_alloy::{
     TempoNetwork,
     provider::ext::TempoProviderExt as _,
@@ -286,17 +286,27 @@ where
                 let (evm_config, recorder) = eth_api.evm_config().with_l1_storage_recorder();
                 let block_executor = evm_config.executor(&mut db);
                 let mode = ExecutionWitnessMode::default();
-                let mut witness_record = ExecutionWitnessRecord::default();
+                let mut witness = None;
 
                 let _ = block_executor
                     .execute_with_state_closure(&block, |statedb: &State<_>| {
-                        witness_record.record_executed_state(statedb, mode);
-                        record_block_hash_storage_proofs(&mut witness_record, statedb);
+                        let mut additional_state = HashedPostState::default();
+                        record_block_hash_storage_proofs(&mut additional_state, statedb);
+                        witness = Some(
+                            ExecutionWitnessRecord::new(statedb)
+                                .with_additional_state(additional_state)
+                                .into_execution_witness(
+                                    &statedb.database.database.0,
+                                    eth_api.provider(),
+                                    block_number,
+                                    mode,
+                                ),
+                        );
                     })
                     .map_err(|error| EthApiError::Internal(error.into()))?;
 
-                let witness = witness_record
-                    .into_execution_witness(&db.database.0, eth_api.provider(), block_number, mode)
+                let witness = witness
+                    .expect("state closure is called after successful execution")
                     .map_err(EthApiError::from)?;
                 Ok(ZoneExecutionWitness {
                     execution_witness: witness,
@@ -320,17 +330,16 @@ where
 /// Reth records these reads in REVM's block-hash cache and normally proves them with ancestor
 /// headers. Zones already commit the EIP-2935 history contract in state, so adding the matching
 /// storage targets lets the SPF authenticate the same values against the parent state root.
-fn record_block_hash_storage_proofs<DB>(witness: &mut ExecutionWitnessRecord, state: &State<DB>) {
+fn record_block_hash_storage_proofs<DB>(additional_state: &mut HashedPostState, state: &State<DB>) {
     let block_hashes = state.block_hashes.iter().collect::<Vec<_>>();
     if block_hashes.is_empty() {
         return;
     }
 
-    let history_storage = witness
-        .hashed_state
+    let history_storage = additional_state
         .storages
         .entry(keccak256(HISTORY_STORAGE_ADDRESS))
-        .or_insert_with(|| HashedStorage::new(false));
+        .or_default();
     for (number, hash) in block_hashes {
         let slot = U256::from(number % HISTORY_SERVE_WINDOW as u64);
         history_storage.storage.insert(
@@ -1187,6 +1196,7 @@ where
     fn ws_subscribe_logs(&self, mut filter: Filter, auth: AuthContext) -> BoxWsSubscriptionFut<'_> {
         Box::pin(async move {
             let provider = self.eth.api.provider().clone();
+            let api = self.eth.api.clone();
             let caller = auth.caller;
 
             let zone_tokens = self.zone_tokens();
@@ -1195,18 +1205,36 @@ where
 
             let stream = provider
                 .canonical_state_stream()
-                .flat_map(|canon_state| futures::stream::iter(canon_state.block_receipts()))
-                .flat_map(move |(block_receipts, removed)| {
-                    let all_logs = logs_utils::matching_block_logs_with_tx_hashes(
-                        &filter,
-                        block_receipts.block,
-                        block_receipts.timestamp,
-                        block_receipts
-                            .tx_receipts
-                            .iter()
-                            .map(|(tx, receipt)| (*tx, receipt)),
-                        removed,
-                    );
+                .flat_map(move |canon_state| {
+                    let reverted_chains = canon_state.reverted();
+                    let committed_chain = canon_state.committed();
+                    let reverted = reverted_chains.iter().flat_map(|chain| {
+                        chain
+                            .blocks_and_receipts()
+                            .map(|(block, receipts)| (block, receipts, true))
+                    });
+                    let committed = committed_chain
+                        .blocks_and_receipts()
+                        .map(|(block, receipts)| (block, receipts, false));
+                    let mut all_logs = Vec::new();
+
+                    for (block, receipts, removed) in reverted.chain(committed) {
+                        match logs_utils::matching_block_logs_with_tx_hashes(
+                            api.converter(),
+                            &filter,
+                            block.sealed_header(),
+                            block
+                                .transactions_recovered()
+                                .zip(receipts.iter())
+                                .map(|(tx, receipt)| (*tx.tx_hash(), receipt)),
+                            removed,
+                        ) {
+                            Ok(logs) => all_logs.extend(logs),
+                            Err(error) => {
+                                tracing::error!(target: "rpc", %error, "Failed to convert logs");
+                            }
+                        }
+                    }
                     futures::stream::iter(all_logs)
                 });
 
@@ -1484,12 +1512,11 @@ mod tests {
             .with_database(revm::database::EmptyDB::default())
             .build();
         state.block_hashes.insert(number, hash);
-        let mut witness = ExecutionWitnessRecord::default();
+        let mut additional_state = HashedPostState::default();
 
-        record_block_hash_storage_proofs(&mut witness, &state);
+        record_block_hash_storage_proofs(&mut additional_state, &state);
 
-        let storage = witness
-            .hashed_state
+        let storage = additional_state
             .storages
             .get(&keccak256(HISTORY_STORAGE_ADDRESS))
             .unwrap();

crates/spf/src/mpt.rs

@@ -130,10 +130,9 @@ impl StatelessSparseTrie {
             for (hashed_address, storage) in storage_updates {
                 let current_account = self.trie_account(hashed_address)?;
                 let has_revealed_storage = self.inner.storage_trie_ref(&hashed_address).is_some();
-                if !storage.wiped
-                    && current_account
-                        .as_ref()
-                        .is_some_and(|account| account.storage_root != EMPTY_ROOT_HASH)
+                if current_account
+                    .as_ref()
+                    .is_some_and(|account| account.storage_root != EMPTY_ROOT_HASH)
                     && !has_revealed_storage
                 {
                     return Err(StatelessSparseTrieError::IncompleteStateUpdate);
@@ -143,11 +142,6 @@ impl StatelessSparseTrie {
                     .inner
                     .take_storage_trie(&hashed_address)
                     .unwrap_or_else(RevealableSparseTrie::revealed_empty);
-                if storage.wiped {
-                    storage_trie
-                        .wipe()
-                        .map_err(|_| StatelessSparseTrieError::InvalidSparseTrie)?;
-                }
 
                 let mut updates = storage
                     .storage