feat(bundles): Support backrun bundles execution

crates/op-rbuilder/src/args/op.rs

@@ -51,12 +51,10 @@ pub struct OpRbuilderArgs {
     /// Whether to enable TIPS Resource Metering
     #[arg(long = "builder.enable-resource-metering", default_value = "false")]
     pub enable_resource_metering: bool,
-    /// Whether to enable TIPS Resource Metering
-    #[arg(
-        long = "builder.resource-metering-buffer-size",
-        default_value = "10000"
-    )]
-    pub resource_metering_buffer_size: usize,
+
+    /// Buffer size for tx data store (LRU eviction when full)
+    #[arg(long = "builder.tx-data-store-buffer-size", default_value = "10000")]
+    pub tx_data_store_buffer_size: usize,
 
     /// Path to builder playgorund to automatically start up the node connected to it
     #[arg(

crates/op-rbuilder/src/builders/context.rs

@@ -43,9 +43,9 @@ use crate::{
     gas_limiter::AddressGasLimiter,
     metrics::OpRBuilderMetrics,
     primitives::reth::{ExecutionInfo, TxnExecutionResult},
-    resource_metering::ResourceMetering,
     traits::PayloadTxsBounds,
     tx::MaybeRevertingTransaction,
+    tx_data_store::TxDataStore,
     tx_signer::Signer,
 };
 
@@ -78,8 +78,8 @@ pub struct OpPayloadBuilderCtx<ExtraCtx: Debug + Default = ()> {
     pub max_gas_per_txn: Option<u64>,
     /// Rate limiting based on gas. This is an optional feature.
     pub address_gas_limiter: AddressGasLimiter,
-    /// Per transaction resource metering information
-    pub resource_metering: ResourceMetering,
+    /// Unified transaction data store (backrun bundles + resource metering)
+    pub tx_data_store: TxDataStore,
 }
 
 impl<ExtraCtx: Debug + Default> OpPayloadBuilderCtx<ExtraCtx> {
@@ -445,7 +445,7 @@ impl<ExtraCtx: Debug + Default> OpPayloadBuilderCtx<ExtraCtx> {
 
             num_txs_considered += 1;
 
-            let _resource_usage = self.resource_metering.get(&tx_hash);
+            let _resource_usage = self.tx_data_store.get_metering(&tx_hash);
 
             // TODO: ideally we should get this from the txpool stream
             if let Some(conditional) = conditional
@@ -543,7 +543,8 @@ impl<ExtraCtx: Debug + Default> OpPayloadBuilderCtx<ExtraCtx> {
                 continue;
             }
 
-            if result.is_success() {
+            let is_success = result.is_success();
+            if is_success {
                 log_txn(TxnExecutionResult::Success);
                 num_txs_simulated_success += 1;
                 self.metrics.successful_tx_gas_used.record(gas_used as f64);
@@ -600,6 +601,138 @@ impl<ExtraCtx: Debug + Default> OpPayloadBuilderCtx<ExtraCtx> {
             // append sender and transaction to the respective lists
             info.executed_senders.push(tx.signer());
             info.executed_transactions.push(tx.into_inner());
+
+            if is_success
+                && let Some(backrun_bundles) = self.tx_data_store.get_backrun_bundles(&tx_hash)
+            {
+                self.metrics.backrun_target_txs_found_total.increment(1);
+                let backrun_start_time = Instant::now();
+
+                // Bundles are pre-sorted by total_priority_fee (descending) from the store
+                'bundle_loop: for stored_bundle in backrun_bundles {
+                    debug!(
+                        target: "payload_builder",
+                        message = "Executing backrun bundle",
+                        tx_hash = ?tx_hash,
+                        bundle_id = ?stored_bundle.bundle_id,
+                        tx_count = stored_bundle.backrun_txs.len(),
+                    );
+
+                    let total_effective_tip: u128 = stored_bundle
+                        .backrun_txs
+                        .iter()
+                        .map(|tx| tx.effective_tip_per_gas(base_fee).unwrap_or(0))
+                        .sum();
+                    if total_effective_tip < miner_fee {
+                        self.metrics
+                            .backrun_bundles_rejected_low_fee_total
+                            .increment(1);
+                        info!(
+                            target: "payload_builder",
+                            bundle_id = ?stored_bundle.bundle_id,
+                            target_fee = miner_fee,
+                            total_effective_tip = total_effective_tip,
+                            "Backrun bundle rejected: total effective tip below target tx"
+                        );
+                        break 'bundle_loop;
+                    }
+
+                    let total_backrun_gas: u64 = stored_bundle
+                        .backrun_txs
+                        .iter()
+                        .map(|tx| tx.gas_limit())
+                        .sum();
+                    let total_backrun_da_size: u64 = stored_bundle
+                        .backrun_txs
+                        .iter()
+                        .map(|tx| tx.encoded_2718().len() as u64)
+                        .sum();
+
+                    if let Err(result) = info.is_tx_over_limits(
+                        total_backrun_da_size,
+                        block_gas_limit,
+                        tx_da_limit,
+                        block_da_limit,
+                        total_backrun_gas,
+                        info.da_footprint_scalar,
+                        block_da_footprint_limit,
+                    ) {
+                        self.metrics
+                            .backrun_bundles_rejected_over_limits_total
+                            .increment(1);
+                        info!(
+                            target: "payload_builder",
+                            bundle_id = ?stored_bundle.bundle_id,
+                            result = ?result,
+                            "Backrun bundle rejected: exceeds block limits"
+                        );
+                        continue 'bundle_loop;
+                    }
+
+                    // All-or-nothing: simulate all txs first, only commit if all succeed
+                    let mut pending_results = Vec::with_capacity(stored_bundle.backrun_txs.len());
+
+                    for backrun_tx in &stored_bundle.backrun_txs {
+                        let ResultAndState { result, state } = match evm.transact(backrun_tx) {
+                            Ok(res) => res,
+                            Err(err) => {
+                                return Err(PayloadBuilderError::evm(err));
+                            }
+                        };
+
+                        if !result.is_success() {
+                            self.metrics.backrun_bundles_reverted_total.increment(1);
+                            info!(
+                                target: "payload_builder",
+                                target_tx = ?tx_hash,
+                                failed_tx = ?backrun_tx.tx_hash(),
+                                bundle_id = ?stored_bundle.bundle_id,
+                                gas_used = result.gas_used(),
+                                "Backrun bundle reverted (all-or-nothing)"
+                            );
+                            continue 'bundle_loop;
+                        }
+
+                        pending_results.push((backrun_tx, result, state));
+                    }
+
+                    for (backrun_tx, result, state) in pending_results {
+                        let backrun_gas_used = result.gas_used();
+
+                        info.cumulative_gas_used += backrun_gas_used;
+                        info.cumulative_da_bytes_used += backrun_tx.encoded_2718().len() as u64;
+
+                        let ctx = ReceiptBuilderCtx {
+                            tx: backrun_tx.inner(),
+                            evm: &evm,
+                            result,
+                            state: &state,
+                            cumulative_gas_used: info.cumulative_gas_used,
+                        };
+                        info.receipts.push(self.build_receipt(ctx, None));
+
+                        evm.db_mut().commit(state);
+
+                        let miner_fee = backrun_tx
+                            .effective_tip_per_gas(base_fee)
+                            .expect("fee is always valid; execution succeeded");
+                        info.total_fees += U256::from(miner_fee) * U256::from(backrun_gas_used);
+
+                        info.executed_senders.push(backrun_tx.signer());
+                        info.executed_transactions
+                            .push(backrun_tx.clone().into_inner());
+                    }
+
+                    self.metrics.backrun_bundles_landed_total.increment(1);
+                }
+
+                self.metrics
+                    .backrun_bundle_execution_duration
+                    .record(backrun_start_time.elapsed());
+
+                // Remove the target tx from the backrun bundle store as already executed
+                self.tx_data_store.remove_backrun_bundles(&tx_hash);
+            }
         }
 
         let payload_transaction_simulation_time = execute_txs_start_time.elapsed();

crates/op-rbuilder/src/builders/flashblocks/ctx.rs

@@ -2,8 +2,8 @@ use crate::{
     builders::{BuilderConfig, OpPayloadBuilderCtx, flashblocks::FlashblocksConfig},
     gas_limiter::{AddressGasLimiter, args::GasLimiterArgs},
     metrics::OpRBuilderMetrics,
-    resource_metering::ResourceMetering,
     traits::ClientBounds,
+    tx_data_store::TxDataStore,
 };
 use op_revm::OpSpecId;
 use reth_basic_payload_builder::PayloadConfig;
@@ -30,8 +30,8 @@ pub(super) struct OpPayloadSyncerCtx {
     max_gas_per_txn: Option<u64>,
     /// The metrics for the builder
     metrics: Arc<OpRBuilderMetrics>,
-    /// Resource metering tracking
-    resource_metering: ResourceMetering,
+    /// Unified transaction data store (backrun bundles + resource metering)
+    tx_data_store: TxDataStore,
 }
 
 impl OpPayloadSyncerCtx {
@@ -51,7 +51,7 @@ impl OpPayloadSyncerCtx {
             chain_spec,
             max_gas_per_txn: builder_config.max_gas_per_txn,
             metrics,
-            resource_metering: builder_config.resource_metering,
+            tx_data_store: builder_config.tx_data_store,
         })
     }
 
@@ -84,7 +84,7 @@ impl OpPayloadSyncerCtx {
             extra_ctx: (),
             max_gas_per_txn: self.max_gas_per_txn,
             address_gas_limiter: AddressGasLimiter::new(GasLimiterArgs::default()),
-            resource_metering: self.resource_metering.clone(),
+            tx_data_store: self.tx_data_store.clone(),
         }
     }
 }

crates/op-rbuilder/src/builders/flashblocks/payload.rs

@@ -282,7 +282,7 @@ where
             extra_ctx,
             max_gas_per_txn: self.config.max_gas_per_txn,
             address_gas_limiter: self.address_gas_limiter.clone(),
-            resource_metering: self.config.resource_metering.clone(),
+            tx_data_store: self.config.tx_data_store.clone(),
         })
     }
 

crates/op-rbuilder/src/builders/mod.rs

@@ -21,7 +21,7 @@ mod flashblocks;
 mod generator;
 mod standard;
 
-use crate::resource_metering::ResourceMetering;
+use crate::tx_data_store::TxDataStore;
 pub use builder_tx::{
     BuilderTransactionCtx, BuilderTransactionError, BuilderTransactions, InvalidContractDataError,
     SimulationSuccessResult, get_balance, get_nonce,
@@ -128,8 +128,8 @@ pub struct BuilderConfig<Specific: Clone> {
     /// Address gas limiter stuff
     pub gas_limiter_config: GasLimiterArgs,
 
-    /// Resource metering context
-    pub resource_metering: ResourceMetering,
+    /// Unified transaction data store (backrun bundles + resource metering)
+    pub tx_data_store: TxDataStore,
 }
 
 impl<S: Debug + Clone> core::fmt::Debug for BuilderConfig<S> {
@@ -152,6 +152,7 @@ impl<S: Debug + Clone> core::fmt::Debug for BuilderConfig<S> {
             .field("specific", &self.specific)
             .field("max_gas_per_txn", &self.max_gas_per_txn)
             .field("gas_limiter_config", &self.gas_limiter_config)
+            .field("tx_data_store", &self.tx_data_store)
             .finish()
     }
 }
@@ -170,7 +171,7 @@ impl<S: Default + Clone> Default for BuilderConfig<S> {
             sampling_ratio: 100,
             max_gas_per_txn: None,
             gas_limiter_config: GasLimiterArgs::default(),
-            resource_metering: ResourceMetering::default(),
+            tx_data_store: TxDataStore::default(),
         }
     }
 }
@@ -193,9 +194,9 @@ where
             sampling_ratio: args.telemetry.sampling_ratio,
             max_gas_per_txn: args.max_gas_per_txn,
             gas_limiter_config: args.gas_limiter.clone(),
-            resource_metering: ResourceMetering::new(
+            tx_data_store: TxDataStore::new(
                 args.enable_resource_metering,
-                args.resource_metering_buffer_size,
+                args.tx_data_store_buffer_size,
             ),
             specific: S::try_from(args)?,
         })

crates/op-rbuilder/src/builders/standard/payload.rs

@@ -251,7 +251,7 @@ where
             extra_ctx: Default::default(),
             max_gas_per_txn: self.config.max_gas_per_txn,
             address_gas_limiter: self.address_gas_limiter.clone(),
-            resource_metering: self.config.resource_metering.clone(),
+            tx_data_store: self.config.tx_data_store.clone(),
         };
 
         let builder = OpBuilder::new(best);

crates/op-rbuilder/src/launcher.rs

@@ -7,9 +7,9 @@ use crate::{
     metrics::{VERSION, record_flag_gauge_metrics},
     monitor_tx_pool::monitor_tx_pool,
     primitives::reth::engine_api_builder::OpEngineApiBuilder,
-    resource_metering::{BaseApiExtServer, ResourceMeteringExt},
     revert_protection::{EthApiExtServer, RevertProtectionExt},
     tx::FBPooledTransaction,
+    tx_data_store::{BaseApiExtServer, TxDataStoreExt},
 };
 use core::fmt::Debug;
 use moka::future::Cache;
@@ -110,7 +110,7 @@ where
         let op_node = OpNode::new(rollup_args.clone());
         let reverted_cache = Cache::builder().max_capacity(100).build();
         let reverted_cache_copy = reverted_cache.clone();
-        let resource_metering = builder_config.resource_metering.clone();
+        let tx_data_store = builder_config.tx_data_store.clone();
 
         let mut addons: OpAddOns<
             _,
@@ -166,9 +166,9 @@ where
                         .add_or_replace_configured(revert_protection_ext.into_rpc())?;
                 }
 
-                let resource_metering_ext = ResourceMeteringExt::new(resource_metering);
+                let tx_data_store_ext = TxDataStoreExt::new(tx_data_store);
                 ctx.modules
-                    .add_or_replace_configured(resource_metering_ext.into_rpc())?;
+                    .add_or_replace_configured(tx_data_store_ext.into_rpc())?;
 
                 Ok(())
             })

crates/op-rbuilder/src/lib.rs

@@ -9,10 +9,10 @@ pub mod primitives;
 pub mod revert_protection;
 pub mod traits;
 pub mod tx;
+pub mod tx_data_store;
 pub mod tx_signer;
 
 #[cfg(test)]
 pub mod mock_tx;
-mod resource_metering;
 #[cfg(any(test, feature = "testing"))]
 pub mod tests;

crates/op-rbuilder/src/metrics.rs

@@ -167,6 +167,24 @@ pub struct OpRBuilderMetrics {
     pub metering_unknown_transaction: Counter,
     /// Count of the number of times we were unable to resolve metering information due to locking
     pub metering_locked_transaction: Counter,
+    /// Current number of backrun bundles in store
+    pub backrun_bundles_in_store: Gauge,
+    /// Number of target transactions found with backrun bundles
+    pub backrun_target_txs_found_total: Counter,
+    /// Number of backrun bundles received via RPC
+    pub backrun_bundles_received_total: Counter,
+    /// Number of backrun bundles that reverted during execution (all-or-nothing)
+    pub backrun_bundles_reverted_total: Counter,
+    /// Number of backrun bundles rejected due to priority fee below target tx
+    pub backrun_bundles_rejected_low_fee_total: Counter,
+    /// Number of backrun bundles rejected due to exceeding block limits
+    pub backrun_bundles_rejected_over_limits_total: Counter,
+    /// Number of backrun bundles successfully landed in a block
+    pub backrun_bundles_landed_total: Counter,
+    /// Latency of inserting a backrun bundle into the store
+    pub backrun_bundle_insert_duration: Histogram,
+    /// Duration of executing all backrun bundles for a target transaction
+    pub backrun_bundle_execution_duration: Histogram,
 }
 
 impl OpRBuilderMetrics {