feat: track and optionally enforce resource limits during building

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

@@ -51,7 +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
+    /// Whether to enforce resource metering limits (reject transactions that exceed limits)
+    #[arg(long = "builder.enforce-resource-metering", default_value = "false")]
+    pub enforce_resource_metering: bool,
+    /// Buffer size for resource metering data
     #[arg(
         long = "builder.resource-metering-buffer-size",
         default_value = "10000"

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

@@ -0,0 +1,26 @@
+//! Base-specific builder context.
+
+use super::metrics::BaseMetrics;
+
+/// Base-specific context for payload building.
+/// Add this as a single field to OpPayloadBuilderCtx to minimize diff.
+#[derive(Debug, Default, Clone)]
+pub struct BaseBuilderCtx {
+    /// Block execution time limit in microseconds
+    pub block_execution_time_limit_us: u128,
+    /// Whether to enforce resource metering limits
+    pub enforce_limits: bool,
+    /// Base-specific metrics
+    pub metrics: BaseMetrics,
+}
+
+impl BaseBuilderCtx {
+    /// Create a new BaseBuilderCtx with the given execution time limit.
+    pub fn new(block_execution_time_limit_us: u128, enforce_limits: bool) -> Self {
+        Self {
+            block_execution_time_limit_us,
+            enforce_limits,
+            metrics: Default::default(),
+        }
+    }
+}

crates/op-rbuilder/src/base/execution.rs

@@ -0,0 +1,147 @@
+//! Base-specific execution time tracking and limit checking.
+
+use super::metrics::BaseMetrics;
+use crate::resource_metering::ResourceMetering;
+use alloy_primitives::TxHash;
+use tracing::warn;
+
+/// Base-specific execution state bundled into one type.
+/// Add this as a single field to ExecutionInfo to minimize diff.
+#[derive(Debug, Default, Clone)]
+pub struct BaseExecutionState {
+    pub cumulative_execution_time_us: u128,
+}
+
+/// Base-specific transaction usage bundled into one type.
+#[derive(Debug, Default, Clone, Copy)]
+pub struct BaseTxUsage {
+    pub execution_time_us: u128,
+}
+
+/// Base-specific block limits bundled into one type.
+#[derive(Debug, Clone, Copy)]
+pub struct BaseBlockLimits {
+    pub execution_time_us: u128,
+}
+
+/// Result type for Base-specific limit checks.
+#[derive(Debug)]
+pub enum BaseLimitExceeded {
+    ExecutionTime {
+        tx_hash: TxHash,
+        cumulative_us: u128,
+        tx_us: u128,
+        limit_us: u128,
+        tx_gas: u64,
+        remaining_gas: u64,
+    },
+}
+
+impl BaseLimitExceeded {
+    /// Returns the tx usage that caused the limit to be exceeded.
+    pub fn usage(&self) -> BaseTxUsage {
+        match self {
+            Self::ExecutionTime { tx_us, .. } => BaseTxUsage {
+                execution_time_us: *tx_us,
+            },
+        }
+    }
+
+    /// Log and record metrics for this limit exceeded event.
+    ///
+    /// Only logs/records if this is the first tx to exceed the limit
+    /// (i.e., cumulative was within the limit before this tx).
+    pub fn log_and_record(&self, metrics: &BaseMetrics) {
+        match self {
+            Self::ExecutionTime {
+                tx_hash,
+                cumulative_us,
+                tx_us,
+                limit_us,
+                tx_gas,
+                remaining_gas,
+            } => {
+                // Only log/record for the first tx that exceeds the limit
+                if *cumulative_us > *limit_us {
+                    return;
+                }
+
+                let remaining_us = limit_us.saturating_sub(*cumulative_us);
+                let exceeded_by_us = tx_us.saturating_sub(remaining_us);
+                warn!(
+                    target: "payload_builder",
+                    %tx_hash,
+                    cumulative_us,
+                    tx_us,
+                    limit_us,
+                    remaining_us,
+                    exceeded_by_us,
+                    tx_gas,
+                    remaining_gas,
+                    "Execution time limit exceeded"
+                );
+                metrics.execution_time_limit_exceeded.increment(1);
+                metrics.execution_time_limit_tx_us.record(*tx_us as f64);
+                metrics
+                    .execution_time_limit_remaining_us
+                    .record(remaining_us as f64);
+                metrics
+                    .execution_time_limit_exceeded_by_us
+                    .record(exceeded_by_us as f64);
+                metrics.execution_time_limit_tx_gas.record(*tx_gas as f64);
+                metrics
+                    .execution_time_limit_remaining_gas
+                    .record(*remaining_gas as f64);
+            }
+        }
+    }
+}
+
+impl BaseExecutionState {
+    /// Check if adding a tx would exceed Base-specific limits.
+    /// Call this AFTER the upstream is_tx_over_limits().
+    /// Returns the usage for later recording via `record_tx`.
+    pub fn check_tx(
+        &self,
+        metering: &ResourceMetering,
+        tx_hash: &TxHash,
+        execution_time_limit_us: u128,
+        tx_gas: u64,
+        cumulative_gas_used: u64,
+        block_gas_limit: u64,
+    ) -> Result<BaseTxUsage, BaseLimitExceeded> {
+        let usage = BaseTxUsage::from_metering(metering, tx_hash);
+        let total = self
+            .cumulative_execution_time_us
+            .saturating_add(usage.execution_time_us);
+
+        if total > execution_time_limit_us {
+            let remaining_gas = block_gas_limit.saturating_sub(cumulative_gas_used);
+            return Err(BaseLimitExceeded::ExecutionTime {
+                tx_hash: *tx_hash,
+                cumulative_us: self.cumulative_execution_time_us,
+                tx_us: usage.execution_time_us,
+                limit_us: execution_time_limit_us,
+                tx_gas,
+                remaining_gas,
+            });
+        }
+        Ok(usage)
+    }
+
+    /// Record that a transaction was included.
+    pub fn record_tx(&mut self, usage: &BaseTxUsage) {
+        self.cumulative_execution_time_us += usage.execution_time_us;
+    }
+}
+
+impl BaseTxUsage {
+    /// Get tx execution time from resource metering.
+    pub fn from_metering(metering: &ResourceMetering, tx_hash: &TxHash) -> Self {
+        let execution_time_us = metering
+            .get(tx_hash)
+            .map(|r| r.total_execution_time_us)
+            .unwrap_or(0);
+        Self { execution_time_us }
+    }
+}

crates/op-rbuilder/src/base/flashblocks.rs

@@ -0,0 +1,43 @@
+//! Base-specific flashblocks context.
+
+use super::context::BaseBuilderCtx;
+
+/// Base-specific flashblocks context for per-batch execution time tracking.
+/// Add this as a single field to FlashblocksExtraCtx to minimize diff.
+#[derive(Debug, Default, Clone, Copy)]
+pub struct BaseFlashblocksCtx {
+    /// Total execution time (us) limit for the current flashblock batch
+    pub target_execution_time_us: u128,
+    /// Execution time (us) limit per flashblock batch
+    pub execution_time_per_batch_us: u128,
+    /// Whether to enforce resource metering limits
+    pub enforce_limits: bool,
+}
+
+impl BaseFlashblocksCtx {
+    /// Create a new BaseFlashblocksCtx with the given execution time limit per batch.
+    pub fn new(execution_time_per_batch_us: u128, enforce_limits: bool) -> Self {
+        Self {
+            target_execution_time_us: execution_time_per_batch_us,
+            execution_time_per_batch_us,
+            enforce_limits,
+        }
+    }
+
+    /// Advance to the next batch, updating the target execution time.
+    ///
+    /// Unlike gas and DA, execution time does not carry over to the next batch.
+    pub fn next(self, cumulative_execution_time_us: u128) -> Self {
+        Self {
+            target_execution_time_us: cumulative_execution_time_us
+                + self.execution_time_per_batch_us,
+            ..self
+        }
+    }
+}
+
+impl From<&BaseFlashblocksCtx> for BaseBuilderCtx {
+    fn from(ctx: &BaseFlashblocksCtx) -> Self {
+        BaseBuilderCtx::new(ctx.target_execution_time_us, ctx.enforce_limits)
+    }
+}

crates/op-rbuilder/src/base/metrics.rs

@@ -0,0 +1,24 @@
+//! Base-specific metrics.
+
+use reth_metrics::{
+    Metrics,
+    metrics::{Counter, Histogram},
+};
+
+/// Base-specific metrics for resource metering.
+#[derive(Metrics, Clone)]
+#[metrics(scope = "op_rbuilder_base")]
+pub struct BaseMetrics {
+    /// Count of transactions excluded due to execution time limit
+    pub execution_time_limit_exceeded: Counter,
+    /// Histogram of tx execution time (us) that caused the limit to be exceeded
+    pub execution_time_limit_tx_us: Histogram,
+    /// Histogram of remaining execution time (us) when a tx was excluded
+    pub execution_time_limit_remaining_us: Histogram,
+    /// Histogram of how much the tx exceeded the remaining time (us)
+    pub execution_time_limit_exceeded_by_us: Histogram,
+    /// Histogram of tx gas limit when excluded due to execution time limit
+    pub execution_time_limit_tx_gas: Histogram,
+    /// Histogram of remaining gas when excluded due to execution time limit
+    pub execution_time_limit_remaining_gas: Histogram,
+}

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

@@ -0,0 +1,4 @@
+pub mod context;
+pub mod execution;
+pub mod flashblocks;
+pub mod metrics;

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

@@ -40,6 +40,7 @@ use tokio_util::sync::CancellationToken;
 use tracing::{debug, info, trace};
 
 use crate::{
+    base::context::BaseBuilderCtx,
     gas_limiter::AddressGasLimiter,
     metrics::OpRBuilderMetrics,
     primitives::reth::{ExecutionInfo, TxnExecutionResult},
@@ -80,6 +81,8 @@ pub struct OpPayloadBuilderCtx<ExtraCtx: Debug + Default = ()> {
     pub address_gas_limiter: AddressGasLimiter,
     /// Per transaction resource metering information
     pub resource_metering: ResourceMetering,
+    /// Base-specific builder context
+    pub base_ctx: BaseBuilderCtx,
 }
 
 impl<ExtraCtx: Debug + Default> OpPayloadBuilderCtx<ExtraCtx> {
@@ -381,6 +384,7 @@ impl<ExtraCtx: Debug + Default> OpPayloadBuilderCtx<ExtraCtx> {
     /// Executes the given best transactions and updates the execution info.
     ///
     /// Returns `Ok(Some(())` if the job was cancelled.
+    #[allow(clippy::too_many_arguments)]
     pub(super) fn execute_best_transactions<E: Debug + Default>(
         &self,
         info: &mut ExecutionInfo<E>,
@@ -389,6 +393,7 @@ impl<ExtraCtx: Debug + Default> OpPayloadBuilderCtx<ExtraCtx> {
         block_gas_limit: u64,
         block_da_limit: Option<u64>,
         block_da_footprint_limit: Option<u64>,
+        base_ctx: &BaseBuilderCtx,
     ) -> Result<Option<()>, PayloadBuilderError> {
         let execute_txs_start_time = Instant::now();
         let mut num_txs_considered = 0;
@@ -445,8 +450,6 @@ impl<ExtraCtx: Debug + Default> OpPayloadBuilderCtx<ExtraCtx> {
 
             num_txs_considered += 1;
 
-            let _resource_usage = self.resource_metering.get(&tx_hash);
-
             // TODO: ideally we should get this from the txpool stream
             if let Some(conditional) = conditional
                 && !conditional.matches_block_attributes(&block_attr)
@@ -486,6 +489,26 @@ impl<ExtraCtx: Debug + Default> OpPayloadBuilderCtx<ExtraCtx> {
                 continue;
             }
 
+            // Base addition: execution time limit check
+            let base_usage = match info.base_state.check_tx(
+                &self.resource_metering,
+                &tx_hash,
+                base_ctx.block_execution_time_limit_us,
+                tx.gas_limit(),
+                info.cumulative_gas_used,
+                block_gas_limit,
+            ) {
+                Ok(usage) => usage,
+                Err(exceeded) => {
+                    exceeded.log_and_record(&self.base_ctx.metrics);
+                    if self.base_ctx.enforce_limits {
+                        best_txs.mark_invalid(tx.signer(), tx.nonce());
+                        continue;
+                    }
+                    exceeded.usage()
+                }
+            };
+
             // A sequencer's block should never contain blob or deposit transactions from the pool.
             if tx.is_eip4844() || tx.is_deposit() {
                 log_txn(TxnExecutionResult::SequencerTransaction);
@@ -577,6 +600,8 @@ impl<ExtraCtx: Debug + Default> OpPayloadBuilderCtx<ExtraCtx> {
             info.cumulative_gas_used += gas_used;
             // record tx da size
             info.cumulative_da_bytes_used += tx_da_size;
+            // record Base-specific tx execution time
+            info.base_state.record_tx(&base_usage);
 
             // Push transaction changeset and calculate header bloom filter for receipt.
             let ctx = ReceiptBuilderCtx {

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

@@ -1,4 +1,5 @@
 use crate::{
+    base::context::BaseBuilderCtx,
     builders::{BuilderConfig, OpPayloadBuilderCtx, flashblocks::FlashblocksConfig},
     gas_limiter::{AddressGasLimiter, args::GasLimiterArgs},
     metrics::OpRBuilderMetrics,
@@ -32,6 +33,8 @@ pub(super) struct OpPayloadSyncerCtx {
     metrics: Arc<OpRBuilderMetrics>,
     /// Resource metering tracking
     resource_metering: ResourceMetering,
+    /// Base-specific builder context
+    base_ctx: BaseBuilderCtx,
 }
 
 impl OpPayloadSyncerCtx {
@@ -52,6 +55,10 @@ impl OpPayloadSyncerCtx {
             max_gas_per_txn: builder_config.max_gas_per_txn,
             metrics,
             resource_metering: builder_config.resource_metering,
+            base_ctx: BaseBuilderCtx::new(
+                builder_config.block_time.as_micros(),
+                builder_config.enforce_resource_metering,
+            ),
         })
     }
 
@@ -85,6 +92,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(),
+            base_ctx: self.base_ctx.clone(),
         }
     }
 }