Fix database layering and accumulate flashblocks state for metering

Corrects the database architecture to properly layer State -> CacheDB -> StateProviderDatabase,
and ensures both cache and bundle state accumulate across flashblocks for accurate state root
calculations in bundle metering.

Key changes:
- Fix database layering: State now wraps CacheDB instead of CacheDB wrapping State
- Track accumulated bundle state in PendingBlocks alongside cache
- Apply bundle prestate when building state to include previous flashblocks changes
- Use merge_transitions before take_bundle to properly extract accumulated state

Author: niran

Cargo.lock

@@ -73,6 +73,7 @@ reth-db-common = { git = "https://github.com/paradigmxyz/reth", tag = "v1.8.2" }
 # revm
 revm = { version = "29.0.0", default-features = false }
 revm-bytecode = { version = "6.2.2", default-features = false }
+revm-database = { version = "7.0.5", default-features = false }
 
 # alloy
 alloy-primitives = { version = "1.3.1", default-features = false, features = [

Cargo.toml

@@ -29,6 +29,9 @@ reth-primitives.workspace = true
 reth-primitives-traits.workspace = true
 reth-exex.workspace = true
 
+# revm
+revm-database.workspace = true
+
 # alloy
 alloy-primitives.workspace = true
 alloy-eips.workspace = true

crates/flashblocks-rpc/Cargo.toml

@@ -10,7 +10,7 @@ use alloy_rpc_types_eth::{Filter, Header as RPCHeader, Log};
 use eyre::eyre;
 use op_alloy_network::Optimism;
 use op_alloy_rpc_types::{OpTransactionReceipt, Transaction};
-use reth::revm::{db::Cache, state::EvmState};
+use reth::revm::{db::{BundleState, Cache}, state::EvmState};
 use reth_rpc_eth_api::RpcBlock;
 
 use crate::subscription::Flashblock;
@@ -28,6 +28,7 @@ pub struct PendingBlocksBuilder {
     state_overrides: Option<StateOverride>,
 
     db_cache: Cache,
+    bundle_state: BundleState,
 }
 
 impl PendingBlocksBuilder {
@@ -43,6 +44,7 @@ impl PendingBlocksBuilder {
             transaction_state: HashMap::new(),
             state_overrides: None,
             db_cache: Cache::default(),
+            bundle_state: BundleState::default(),
         }
     }
 
@@ -107,6 +109,12 @@ impl PendingBlocksBuilder {
         self
     }
 
+    #[inline]
+    pub(crate) fn with_bundle_state(&mut self, bundle_state: BundleState) -> &Self {
+        self.bundle_state = bundle_state;
+        self
+    }
+
     pub(crate) fn build(self) -> eyre::Result<PendingBlocks> {
         if self.headers.is_empty() {
             return Err(eyre!("missing headers"));
@@ -127,6 +135,7 @@ impl PendingBlocksBuilder {
             transaction_state: self.transaction_state,
             state_overrides: self.state_overrides,
             db_cache: self.db_cache,
+            bundle_state: self.bundle_state,
         })
     }
 }
@@ -145,6 +154,7 @@ pub struct PendingBlocks {
     state_overrides: Option<StateOverride>,
 
     db_cache: Cache,
+    bundle_state: BundleState,
 }
 
 impl PendingBlocks {
@@ -176,6 +186,10 @@ impl PendingBlocks {
         self.db_cache.clone()
     }
 
+    pub fn get_bundle_state(&self) -> BundleState {
+        self.bundle_state.clone()
+    }
+
     pub fn get_transactions_for_block(&self, block_number: BlockNumber) -> Vec<Transaction> {
         self.transactions
             .iter()

crates/flashblocks-rpc/src/pending_blocks.rs

@@ -22,6 +22,7 @@ use reth::providers::{BlockReaderIdExt, StateProviderFactory};
 use reth::revm::context::result::ResultAndState;
 use reth::revm::database::StateProviderDatabase;
 use reth::revm::db::CacheDB;
+use revm_database::states::bundle_state::BundleRetention;
 use reth::revm::{DatabaseCommit, State};
 use reth_evm::{ConfigureEvm, Evm};
 use reth_optimism_chainspec::OpHardforks;
@@ -406,18 +407,30 @@ where
             .client
             .state_by_block_number_or_tag(BlockNumberOrTag::Number(canonical_block))?;
         let state_provider_db = StateProviderDatabase::new(state_provider);
-        let state = State::builder()
-            .with_database(state_provider_db)
-            .with_bundle_update()
-            .build();
         let mut pending_blocks_builder = PendingBlocksBuilder::new();
 
-        let mut db = match &prev_pending_blocks {
+        // Cache reads across flashblocks, accumulating caches from previous
+        // pending blocks if available
+        let cache_db = match &prev_pending_blocks {
             Some(pending_blocks) => CacheDB {
                 cache: pending_blocks.get_db_cache(),
-                db: state,
+                db: state_provider_db,
             },
-            None => CacheDB::new(state),
+            None => CacheDB::new(state_provider_db),
+        };
+
+        // Track state changes across flashblocks, accumulating bundle state
+        // from previous pending blocks if available
+        let mut db = match &prev_pending_blocks {
+            Some(pending_blocks) => State::builder()
+                .with_database(cache_db)
+                .with_bundle_update()
+                .with_bundle_prestate(pending_blocks.get_bundle_state())
+                .build(),
+            None => State::builder()
+                .with_database(cache_db)
+                .with_bundle_update()
+                .build(),
         };
         let mut state_cache_builder = match &prev_pending_blocks {
             Some(pending_blocks) => {
@@ -658,7 +671,9 @@ where
             last_block_header = block.header.clone();
         }
 
-        pending_blocks_builder.with_db_cache(db.cache);
+        db.merge_transitions(BundleRetention::Reverts);
+        pending_blocks_builder.with_bundle_state(db.take_bundle());
+        pending_blocks_builder.with_db_cache(db.database.cache);
         pending_blocks_builder.with_state_overrides(state_cache_builder.build());
         Ok(Some(Arc::new(pending_blocks_builder.build()?)))
     }

crates/flashblocks-rpc/src/state.rs

@@ -40,6 +40,7 @@ base-reth-flashblocks-rpc = { path = "../flashblocks-rpc" }
 
 # revm
 revm.workspace = true
+revm-database.workspace = true
 
 # rpc
 jsonrpsee.workspace = true

crates/metering/Cargo.toml

@@ -3,6 +3,6 @@ mod rpc;
 #[cfg(test)]
 mod tests;
 
-pub use meter::{meter_bundle, MeterBundleOutput};
+pub use meter::{meter_bundle, FlashblocksState, MeterBundleOutput};
 pub use rpc::{MeteringApiImpl, MeteringApiServer};
 pub use tips_core::types::{Bundle, MeterBundleResponse, TransactionResult};

crates/metering/src/lib.rs

@@ -1,7 +1,8 @@
 use alloy_consensus::{transaction::SignerRecoverable, BlockHeader, Transaction as _};
 use alloy_primitives::{B256, U256};
 use eyre::{eyre, Result as EyreResult};
-use reth::revm::db::{Cache, CacheDB, State};
+use reth::revm::db::{BundleState, Cache, CacheDB, State};
+use revm_database::states::bundle_state::BundleRetention;
 use reth_evm::execute::BlockBuilder;
 use reth_evm::ConfigureEvm;
 use reth_optimism_chainspec::OpChainSpec;
@@ -12,6 +13,15 @@ use std::time::Instant;
 
 use crate::TransactionResult;
 
+/// State from pending flashblocks that is used as a base for metering
+#[derive(Debug, Clone)]
+pub struct FlashblocksState {
+    /// The cache of account and storage data
+    pub cache: Cache,
+    /// The accumulated bundle of state changes
+    pub bundle_state: BundleState,
+}
+
 const BLOCK_TIME: u64 = 2; // 2 seconds per block
 
 /// Output from metering a bundle of transactions
@@ -43,36 +53,40 @@ pub fn meter_bundle<SP>(
     decoded_txs: Vec<op_alloy_consensus::OpTxEnvelope>,
     header: &SealedHeader,
     bundle_with_metadata: &tips_core::types::BundleWithMetadata,
-    db_cache: Option<Cache>,
+    flashblocks_state: Option<FlashblocksState>,
 ) -> EyreResult<MeterBundleOutput>
 where
     SP: reth_provider::StateProvider,
 {
     // Get bundle hash from BundleWithMetadata
     let bundle_hash = bundle_with_metadata.bundle_hash();
 
-    // Create state database with optional flashblocks cache
+    // Create state database
     let state_db = reth::revm::database::StateProviderDatabase::new(state_provider);
-    let base_state = State::builder()
-        .with_database(state_db)
-        .with_bundle_update()
-        .build();
 
-    // If we have flashblocks cache, wrap with CacheDB to apply pending changes
-    let cache_db = if let Some(cache) = db_cache {
+    // If we have flashblocks state, apply both cache and bundle prestate
+    let cache_db = if let Some(ref flashblocks) = flashblocks_state {
         CacheDB {
-            cache,
-            db: base_state,
+            cache: flashblocks.cache.clone(),
+            db: state_db,
         }
     } else {
-        CacheDB::new(base_state)
+        CacheDB::new(state_db)
     };
 
-    // Wrap the CacheDB in a State for the EVM builder
-    let mut db = State::builder()
-        .with_database(cache_db)
-        .with_bundle_update()
-        .build();
+    // Wrap the CacheDB in a State to track bundle changes for state root calculation
+    let mut db = if let Some(flashblocks) = flashblocks_state.as_ref() {
+        State::builder()
+            .with_database(cache_db)
+            .with_bundle_update()
+            .with_bundle_prestate(flashblocks.bundle_state.clone())
+            .build()
+    } else {
+        State::builder()
+            .with_database(cache_db)
+            .with_bundle_update()
+            .build()
+    };
 
     // Set up next block attributes
     // Use bundle.min_timestamp if provided, otherwise use header timestamp + BLOCK_TIME
@@ -138,11 +152,14 @@ where
     }
 
     // Calculate state root and measure its calculation time
+    // The bundle already includes flashblocks state if it was provided via with_bundle_prestate
+    db.merge_transitions(BundleRetention::Reverts);
     let bundle = db.take_bundle();
-    let state_provider = db.database.as_ref();
+    let state_provider = db.database.db.as_ref();
     let state_root_start = Instant::now();
-    let hashed_state = state_provider.hashed_post_state(&bundle);
-    let _ = state_provider.state_root_with_updates(hashed_state);
+
+    let hashed_post_state = state_provider.hashed_post_state(&bundle);
+    let _ = state_provider.state_root_with_updates(hashed_post_state);
     let state_root_time = state_root_start.elapsed().as_micros();
 
     let total_execution_time = execution_start.elapsed().as_micros();

crates/metering/src/meter.rs

@@ -152,8 +152,11 @@ where
                 )
             })?;
 
-        // If we have pending flashblocks, get the db_cache to apply state changes
-        let db_cache = pending_blocks.as_ref().map(|pb| pb.get_db_cache());
+        // If we have pending flashblocks, get the state to apply pending changes
+        let flashblocks_state = pending_blocks.as_ref().map(|pb| crate::FlashblocksState {
+            cache: pb.get_db_cache(),
+            bundle_state: pb.get_bundle_state(),
+        });
 
         // Meter bundle using utility function
         let result = meter_bundle(
@@ -162,7 +165,7 @@ where
             decoded_txs,
             &header,
             &bundle_with_metadata,
-            db_cache,
+            flashblocks_state,
         )
         .map_err(|e| {
             error!(error = %e, "Bundle metering failed");

crates/metering/src/rpc.rs

@@ -161,7 +161,7 @@ fn meter_bundle_empty_transactions() -> eyre::Result<()> {
         Vec::new(),
         &harness.header,
         &bundle_with_metadata,
-        None,  // No flashblocks cache in tests
+        None,
     )?;
 
     assert!(output.results.is_empty());
@@ -210,7 +210,7 @@ fn meter_bundle_single_transaction() -> eyre::Result<()> {
         vec![envelope],
         &harness.header,
         &bundle_with_metadata,
-        None,  // No flashblocks cache in tests
+        None,
     )?;
 
     assert_eq!(output.results.len(), 1);
@@ -307,7 +307,7 @@ fn meter_bundle_multiple_transactions() -> eyre::Result<()> {
         vec![envelope_1, envelope_2],
         &harness.header,
         &bundle_with_metadata,
-        None,  // No flashblocks cache in tests
+        None,
     )?;
 
     assert_eq!(output.results.len(), 2);
@@ -396,7 +396,7 @@ fn meter_bundle_state_root_time_invariant() -> eyre::Result<()> {
         vec![envelope],
         &harness.header,
         &bundle_with_metadata,
-        None,  // No flashblocks cache in tests
+        None,
     )?;
 
     // Verify invariant: total execution time must include state root time