feat(bundles): Support backrun bundles execution

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

@@ -58,6 +58,10 @@ pub struct OpRbuilderArgs {
     )]
     pub resource_metering_buffer_size: usize,
 
+    /// Buffer size for backrun bundles (LRU eviction when full)
+    #[arg(long = "builder.backrun-bundle-buffer-size", default_value = "10000")]
+    pub backrun_bundle_buffer_size: usize,
+
     /// Path to builder playgorund to automatically start up the node connected to it
     #[arg(
         long = "builder.playground",

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

@@ -80,6 +80,8 @@ pub struct OpPayloadBuilderCtx<ExtraCtx: Debug + Default = ()> {
     pub address_gas_limiter: AddressGasLimiter,
     /// Per transaction resource metering information
     pub resource_metering: ResourceMetering,
+    /// Backrun bundle store for storing backrun transactions
+    pub backrun_bundle_store: crate::bundles::BackrunBundleStore,
 }
 
 impl<ExtraCtx: Debug + Default> OpPayloadBuilderCtx<ExtraCtx> {
@@ -433,7 +435,7 @@ impl<ExtraCtx: Debug + Default> OpPayloadBuilderCtx<ExtraCtx> {
                 is_bundle_tx && !reverted_hashes.unwrap().contains(&tx_hash);
 
             let log_txn = |result: TxnExecutionResult| {
-                debug!(
+                info!(
                     target: "payload_builder",
                     message = "Considering transaction",
                     tx_hash = ?tx_hash,
@@ -543,7 +545,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 +603,120 @@ 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.backrun_bundle_store.get(&tx_hash) {
+                self.metrics.backrun_target_txs_found_total.increment(1);
+                let backrun_start_time = Instant::now();
+
+                // Pre-compute total fees and sort bundles (descending)
+                let mut bundles_with_fees: Vec<_> = backrun_bundles
+                    .into_iter()
+                    .map(|bundle| {
+                        let total_fee: u128 = bundle
+                            .backrun_txs
+                            .iter()
+                            .map(|tx| tx.effective_tip_per_gas(base_fee).unwrap_or(0))
+                            .sum();
+                        (bundle, total_fee)
+                    })
+                    .collect();
+                bundles_with_fees.sort_by(|a, b| b.1.cmp(&a.1));
+
+                'bundle_loop: for (mut stored_bundle, total_bundle_fee) in bundles_with_fees {
+                    info!(
+                        target: "payload_builder",
+                        message = "Executing backrun bundle",
+                        tx_hash = ?tx_hash,
+                        bundle_id = ?stored_bundle.bundle_id,
+                        tx_count = stored_bundle.backrun_txs.len(),
+                    );
+
+                    // Validate: total bundle priority fee must be >= target tx's priority fee
+                    if total_bundle_fee < 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_bundle_fee = total_bundle_fee,
+                            "Backrun bundle rejected: total priority fee below target tx"
+                        );
+                        continue 'bundle_loop;
+                    }
+
+                    // Sort backrun txs by priority fee (descending)
+                    stored_bundle.backrun_txs.sort_unstable_by(|a, b| {
+                        let a_tip = a.effective_tip_per_gas(base_fee).unwrap_or(0);
+                        let b_tip = b.effective_tip_per_gas(base_fee).unwrap_or(0);
+                        b_tip.cmp(&a_tip)
+                    });
+
+                    // 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.backrun_bundle_store.remove(&tx_hash);
+            }
         }
 
         let payload_transaction_simulation_time = execute_txs_start_time.elapsed();

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

@@ -1,5 +1,6 @@
 use crate::{
     builders::{BuilderConfig, OpPayloadBuilderCtx, flashblocks::FlashblocksConfig},
+    bundles::BackrunBundleStore,
     gas_limiter::{AddressGasLimiter, args::GasLimiterArgs},
     metrics::OpRBuilderMetrics,
     resource_metering::ResourceMetering,
@@ -32,6 +33,8 @@ pub(super) struct OpPayloadSyncerCtx {
     metrics: Arc<OpRBuilderMetrics>,
     /// Resource metering tracking
     resource_metering: ResourceMetering,
+    /// Backrun bundle store
+    backrun_bundle_store: BackrunBundleStore,
 }
 
 impl OpPayloadSyncerCtx {
@@ -52,6 +55,7 @@ impl OpPayloadSyncerCtx {
             max_gas_per_txn: builder_config.max_gas_per_txn,
             metrics,
             resource_metering: builder_config.resource_metering,
+            backrun_bundle_store: builder_config.backrun_bundle_store,
         })
     }
 
@@ -85,6 +89,7 @@ impl OpPayloadSyncerCtx {
             max_gas_per_txn: self.max_gas_per_txn,
             address_gas_limiter: AddressGasLimiter::new(GasLimiterArgs::default()),
             resource_metering: self.resource_metering.clone(),
+            backrun_bundle_store: self.backrun_bundle_store.clone(),
         }
     }
 }

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

@@ -283,6 +283,7 @@ where
             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(),
+            backrun_bundle_store: self.config.backrun_bundle_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::{bundles::BackrunBundleStore, resource_metering::ResourceMetering};
 pub use builder_tx::{
     BuilderTransactionCtx, BuilderTransactionError, BuilderTransactions, InvalidContractDataError,
     SimulationSuccessResult, get_balance, get_nonce,
@@ -130,6 +130,9 @@ pub struct BuilderConfig<Specific: Clone> {
 
     /// Resource metering context
     pub resource_metering: ResourceMetering,
+
+    /// Backrun bundle store for storing backrun transactions
+    pub backrun_bundle_store: BackrunBundleStore,
 }
 
 impl<S: Debug + Clone> core::fmt::Debug for BuilderConfig<S> {
@@ -152,6 +155,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("backrun_bundle_store", &self.backrun_bundle_store)
             .finish()
     }
 }
@@ -171,6 +175,7 @@ impl<S: Default + Clone> Default for BuilderConfig<S> {
             max_gas_per_txn: None,
             gas_limiter_config: GasLimiterArgs::default(),
             resource_metering: ResourceMetering::default(),
+            backrun_bundle_store: BackrunBundleStore::default(),
         }
     }
 }
@@ -197,6 +202,7 @@ where
                 args.enable_resource_metering,
                 args.resource_metering_buffer_size,
             ),
+            backrun_bundle_store: BackrunBundleStore::new(args.backrun_bundle_buffer_size),
             specific: S::try_from(args)?,
         })
     }

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

@@ -252,6 +252,7 @@ where
             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(),
+            backrun_bundle_store: self.config.backrun_bundle_store.clone(),
         };
 
         let builder = OpBuilder::new(best);