Use pending flashblocks state for bundle metering

Integrate flashblocks state into metering to execute bundles on top of
pending flashblock state rather than canonical block state. This ensures
metered gas usage and execution time accurately reflect the effects of
pending transactions (nonces, balances, storage, code changes).

Implementation:
- Add flashblocks-rpc dependency to metering crate
- Update meter_bundle() to accept optional db_cache parameter
- Implement three-layer state architecture:
  1. StateProviderDatabase (canonical block base state)
  2. CacheDB (applies flashblock pending changes via cache)
  3. State wrapper (for EVM builder compatibility)
- Update MeteringApiImpl to accept FlashblocksState
- Get pending blocks and db_cache from flashblocks when available
- Fall back to canonical block state when no flashblocks available
- Update response to include flashblock_index in logs
- Require flashblocks to be enabled for metering RPC
- Update all tests to pass FlashblocksState parameter

The metering RPC now uses the same state as flashblocks eth_call,
ensuring consistent simulation results.

Author: niran

Cargo.lock

@@ -35,6 +35,9 @@ alloy-eips.workspace = true
 # op-alloy
 op-alloy-consensus.workspace = true
 
+# base
+base-reth-flashblocks-rpc = { path = "../flashblocks-rpc" }
+
 # revm
 revm.workspace = true
 

crates/metering/Cargo.toml

@@ -11,7 +11,7 @@ Simulates a bundle of transactions, providing gas usage and execution time metri
 The method accepts a Bundle object with the following fields:
 
 - `txs`: Array of signed, RLP-encoded transactions (hex strings with 0x prefix)
-- `block_number`: Target block number for bundle validity (note: simulation always uses the latest available block state)
+- `block_number`: Target block number for bundle validity (note: simulation uses pending flashblocks state when available, otherwise latest canonical block)
 - `min_timestamp` (optional): Minimum timestamp for bundle validity (also used as simulation timestamp if provided)
 - `max_timestamp` (optional): Maximum timestamp for bundle validity
 - `reverting_tx_hashes` (optional): Array of transaction hashes allowed to revert
@@ -26,7 +26,7 @@ The method accepts a Bundle object with the following fields:
 - `coinbaseDiff`: Total gas fees paid
 - `ethSentToCoinbase`: ETH sent directly to coinbase
 - `gasFees`: Total gas fees
-- `stateBlockNumber`: Block number used for state (always the latest available block)
+- `stateBlockNumber`: Block number used for state (latest flashblock if pending flashblocks exist, otherwise latest canonical block)
 - `totalGasUsed`: Total gas consumed
 - `totalExecutionTimeUs`: Total execution time (μs)
 - `results`: Array of per-transaction results:

crates/metering/README.md

@@ -1,7 +1,7 @@
 use alloy_consensus::{transaction::SignerRecoverable, BlockHeader, Transaction as _};
 use alloy_primitives::{B256, U256};
 use eyre::{eyre, Result as EyreResult};
-use reth::revm::db::State;
+use reth::revm::db::{Cache, CacheDB, State};
 use reth_evm::execute::BlockBuilder;
 use reth_evm::ConfigureEvm;
 use reth_optimism_chainspec::OpChainSpec;
@@ -43,20 +43,37 @@ 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>,
 ) -> EyreResult<MeterBundleOutput>
 where
     SP: reth_provider::StateProvider,
 {
     // Get bundle hash from BundleWithMetadata
     let bundle_hash = bundle_with_metadata.bundle_hash();
 
-    // Create state database
+    // Create state database with optional flashblocks cache
     let state_db = reth::revm::database::StateProviderDatabase::new(state_provider);
-    let mut db = State::builder()
+    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 {
+        CacheDB {
+            cache,
+            db: base_state,
+        }
+    } else {
+        CacheDB::new(base_state)
+    };
+
+    // Wrap the CacheDB in a State for the EVM builder
+    let mut db = 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
     let timestamp = bundle_with_metadata

crates/metering/src/meter.rs

@@ -1,14 +1,16 @@
-use alloy_consensus::Header;
+use alloy_consensus::{Header, Sealed};
 use alloy_eips::eip2718::Decodable2718;
-use alloy_eips::BlockNumberOrTag;
 use alloy_primitives::U256;
+use base_reth_flashblocks_rpc::rpc::{FlashblocksAPI, PendingBlocksAPI};
 use jsonrpsee::{
     core::{async_trait, RpcResult},
     proc_macros::rpc,
 };
 use reth::providers::BlockReaderIdExt;
 use reth_optimism_chainspec::OpChainSpec;
+use reth_primitives_traits::SealedHeader;
 use reth_provider::{ChainSpecProvider, StateProviderFactory};
+use std::sync::Arc;
 use tips_core::types::{Bundle, BundleWithMetadata, MeterBundleResponse};
 use tracing::{error, info};
 
@@ -23,25 +25,30 @@ pub trait MeteringApi {
 }
 
 /// Implementation of the metering RPC API
-pub struct MeteringApiImpl<Provider> {
+pub struct MeteringApiImpl<Provider, FB> {
     provider: Provider,
+    flashblocks_state: Arc<FB>,
 }
 
-impl<Provider> MeteringApiImpl<Provider>
+impl<Provider, FB> MeteringApiImpl<Provider, FB>
 where
     Provider: StateProviderFactory
         + ChainSpecProvider<ChainSpec = OpChainSpec>
         + BlockReaderIdExt<Header = Header>
         + Clone,
+    FB: FlashblocksAPI,
 {
     /// Creates a new instance of MeteringApi
-    pub fn new(provider: Provider) -> Self {
-        Self { provider }
+    pub fn new(provider: Provider, flashblocks_state: Arc<FB>) -> Self {
+        Self {
+            provider,
+            flashblocks_state,
+        }
     }
 }
 
 #[async_trait]
-impl<Provider> MeteringApiServer for MeteringApiImpl<Provider>
+impl<Provider, FB> MeteringApiServer for MeteringApiImpl<Provider, FB>
 where
     Provider: StateProviderFactory
         + ChainSpecProvider<ChainSpec = OpChainSpec>
@@ -50,6 +57,7 @@ where
         + Send
         + Sync
         + 'static,
+    FB: FlashblocksAPI + Send + Sync + 'static,
 {
     async fn meter_bundle(&self, bundle: Bundle) -> RpcResult<MeterBundleResponse> {
         info!(
@@ -58,24 +66,54 @@ where
             "Starting bundle metering"
         );
 
-        // Get the latest header
-        let header = self
-            .provider
-            .sealed_header_by_number_or_tag(BlockNumberOrTag::Latest)
-            .map_err(|e| {
-                jsonrpsee::types::ErrorObjectOwned::owned(
-                    jsonrpsee::types::ErrorCode::InternalError.code(),
-                    format!("Failed to get latest header: {}", e),
-                    None::<()>,
-                )
-            })?
-            .ok_or_else(|| {
-                jsonrpsee::types::ErrorObjectOwned::owned(
-                    jsonrpsee::types::ErrorCode::InternalError.code(),
-                    "Latest block not found".to_string(),
-                    None::<()>,
-                )
-            })?;
+        // Get pending flashblocks state
+        let pending_blocks = self.flashblocks_state.get_pending_blocks();
+
+        // Get header and flashblock index from pending blocks
+        // If no pending blocks exist, fall back to latest canonical block
+        let (header, flashblock_index, canonical_block_number) = if let Some(pb) = pending_blocks.as_ref() {
+            let latest_header: Sealed<Header> = pb.latest_header();
+            let flashblock_index = pb.latest_flashblock_index();
+            let canonical_block_number = pb.canonical_block_number();
+
+            info!(
+                latest_block = latest_header.number,
+                canonical_block = %canonical_block_number,
+                flashblock_index = flashblock_index,
+                "Using latest flashblock state for metering"
+            );
+
+            // Convert Sealed<Header> to SealedHeader
+            let sealed_header = SealedHeader::new(latest_header.inner().clone(), latest_header.hash());
+            (sealed_header, flashblock_index, canonical_block_number)
+        } else {
+            // No pending blocks, use latest canonical block
+            let canonical_block_number = pending_blocks.get_canonical_block_number();
+            let header = self
+                .provider
+                .sealed_header_by_number_or_tag(canonical_block_number)
+                .map_err(|e| {
+                    jsonrpsee::types::ErrorObjectOwned::owned(
+                        jsonrpsee::types::ErrorCode::InternalError.code(),
+                        format!("Failed to get canonical block header: {}", e),
+                        None::<()>,
+                    )
+                })?
+                .ok_or_else(|| {
+                    jsonrpsee::types::ErrorObjectOwned::owned(
+                        jsonrpsee::types::ErrorCode::InternalError.code(),
+                        "Canonical block not found".to_string(),
+                        None::<()>,
+                    )
+                })?;
+
+            info!(
+                canonical_block = header.number,
+                "No flashblocks available, using canonical block state for metering"
+            );
+
+            (header, 0, canonical_block_number)
+        };
 
         // Manually decode transactions to OpTxEnvelope (op-alloy 0.20) instead of using
         // BundleWithMetadata.transactions() which returns op-alloy 0.21 types incompatible with reth.
@@ -101,10 +139,10 @@ where
             )
         })?;
 
-        // Get state provider for the block
+        // Get state provider for the canonical block
         let state_provider = self
             .provider
-            .state_by_block_hash(header.hash())
+            .state_by_block_number_or_tag(canonical_block_number)
             .map_err(|e| {
                 error!(error = %e, "Failed to get state provider");
                 jsonrpsee::types::ErrorObjectOwned::owned(
@@ -114,13 +152,17 @@ 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());
+
         // Meter bundle using utility function
         let result = meter_bundle(
             state_provider,
             self.provider.chain_spec().clone(),
             decoded_txs,
             &header,
             &bundle_with_metadata,
+            db_cache,
         )
         .map_err(|e| {
             error!(error = %e, "Bundle metering failed");
@@ -144,9 +186,13 @@ where
             total_gas_used = result.total_gas_used,
             total_execution_time_us = result.total_execution_time_us,
             state_root_time_us = result.state_root_time_us,
+            state_block_number = header.number,
+            flashblock_index = flashblock_index,
             "Bundle metering completed successfully"
         );
 
+        // TODO: Add flashblock_index to MeterBundleResponse in tips-core
+        // The response should indicate both the canonical block number and the flashblock index
         Ok(MeterBundleResponse {
             bundle_gas_price,
             bundle_hash: result.bundle_hash,

crates/metering/src/rpc.rs

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

crates/metering/src/tests/meter.rs

@@ -6,6 +6,7 @@ mod tests {
     use alloy_primitives::bytes;
     use alloy_primitives::{address, b256, Bytes, U256};
     use alloy_rpc_client::RpcClient;
+    use base_reth_flashblocks_rpc::state::FlashblocksState;
     use op_alloy_consensus::OpTxEnvelope;
     use reth::args::{DiscoveryArgs, NetworkArgs, RpcServerArgs};
     use reth::builder::{Node, NodeBuilder, NodeConfig, NodeHandle};
@@ -91,7 +92,9 @@ mod tests {
             .with_components(node.components_builder())
             .with_add_ons(node.add_ons())
             .extend_rpc_modules(move |ctx| {
-                let metering_api = MeteringApiImpl::new(ctx.provider().clone());
+                // Create a FlashblocksState without starting it (no pending blocks for testing)
+                let flashblocks_state = Arc::new(FlashblocksState::new(ctx.provider().clone()));
+                let metering_api = MeteringApiImpl::new(ctx.provider().clone(), flashblocks_state);
                 ctx.modules.merge_configured(metering_api.into_rpc())?;
                 Ok(())
             })

crates/metering/src/tests/rpc.rs

@@ -138,12 +138,6 @@ fn main() {
                     }
                 })
                 .extend_rpc_modules(move |ctx| {
-                    if metering_enabled {
-                        info!(message = "Starting Metering RPC");
-                        let metering_api = MeteringApiImpl::new(ctx.provider().clone());
-                        ctx.modules.merge_configured(metering_api.into_rpc())?;
-                    }
-
                     if flashblocks_enabled {
                         info!(message = "Starting Flashblocks");
 
@@ -164,12 +158,23 @@ fn main() {
                         let api_ext = EthApiExt::new(
                             ctx.registry.eth_api().clone(),
                             ctx.registry.eth_handlers().filter.clone(),
-                            fb,
+                            fb.clone(),
                         );
 
                         ctx.modules.replace_configured(api_ext.into_rpc())?;
+
+                        if metering_enabled {
+                            info!(message = "Starting Metering RPC with Flashblocks state");
+                            let metering_api = MeteringApiImpl::new(ctx.provider().clone(), fb);
+                            ctx.modules.merge_configured(metering_api.into_rpc())?;
+                        }
                     } else {
                         info!(message = "flashblocks integration is disabled");
+                        if metering_enabled {
+                            return Err(eyre::eyre!(
+                                "Metering RPC requires flashblocks to be enabled (--websocket-url)"
+                            ));
+                        }
                     }
                     Ok(())
                 })