diff --git a/crates/op-rbuilder/src/builder/context.rs b/crates/op-rbuilder/src/builder/context.rs index 67ac97eeb..cae08abe1 100644 --- a/crates/op-rbuilder/src/builder/context.rs +++ b/crates/op-rbuilder/src/builder/context.rs @@ -38,8 +38,8 @@ use tracing::{debug, info, trace}; use crate::{ backrun_bundle::BackrunBundlesPayloadCtx, evm::OpBlockEvmFactory, - gas_limiter::AddressGasLimiter, - metrics::OpRBuilderMetrics, + gas_limiter::GasLimiters, + metrics::{OpRBuilderMetrics, record_tx_simulation_duration}, primitives::reth::{ExecutionInfo, TxnExecutionResult}, traits::PayloadTxsBounds, }; @@ -66,8 +66,9 @@ pub struct OpPayloadBuilderCtx { pub max_gas_per_txn: Option, /// Maximum cumulative uncompressed (EIP-2718 encoded) block size in bytes. pub max_uncompressed_block_size: Option, - /// Rate limiting based on gas. This is an optional feature. - pub address_gas_limiter: AddressGasLimiter, + /// Per-source gas rate limiters (one per tx source). `None` when the + /// limiter is disabled in config. + pub gas_limiters: Option, /// Backrun bundles context. pub backrun_ctx: BackrunBundlesPayloadCtx, /// Skip reverted txs in subsequent flashblocks @@ -560,9 +561,11 @@ impl OpPayloadBuilderCtx { } }; - self.metrics - .tx_simulation_duration - .record(tx_simulation_start_time.elapsed()); + record_tx_simulation_duration( + tx_simulation_start_time.elapsed(), + is_bundle_tx, + &result, + ); self.metrics.tx_byte_size.record(tx.inner().size() as f64); num_txs_simulated += 1; @@ -583,11 +586,11 @@ impl OpPayloadBuilderCtx { ); } - if self - .address_gas_limiter - .consume_gas(tx.signer(), gas_used) - .is_err() - { + let gas_limiter = self + .gas_limiters + .as_ref() + .map(|l| if is_bundle_tx { &l.bundle } else { &l.mempool }); + if gas_limiter.is_some_and(|l| l.consume_gas(tx.signer(), gas_used).is_err()) { log_txn(TxnExecutionResult::MaxGasUsageExceeded); best_txs.mark_invalid(tx.signer(), tx.nonce()); continue; @@ -834,9 +837,7 @@ impl OpPayloadBuilderCtx { continue; } }; - self.metrics - .tx_simulation_duration - .record(br_simulation_start.elapsed()); + record_tx_simulation_duration(br_simulation_start.elapsed(), true, &br_result); self.metrics .tx_byte_size .record(bundle.backrun_tx.inner().size() as f64); @@ -844,11 +845,11 @@ impl OpPayloadBuilderCtx { let br_gas_used = br_result.gas_used(); - if self - .address_gas_limiter - .consume_gas(bundle.backrun_tx.signer(), br_gas_used) - .is_err() - { + if self.gas_limiters.as_ref().is_some_and(|l| { + l.bundle + .consume_gas(bundle.backrun_tx.signer(), br_gas_used) + .is_err() + }) { log_br_txn(TxnExecutionResult::MaxGasUsageExceeded); continue; } diff --git a/crates/op-rbuilder/src/builder/payload.rs b/crates/op-rbuilder/src/builder/payload.rs index a5cbdf47b..4bea71808 100644 --- a/crates/op-rbuilder/src/builder/payload.rs +++ b/crates/op-rbuilder/src/builder/payload.rs @@ -10,7 +10,7 @@ use crate::{ timing::{FlashblockScheduler, compute_slot_offset_ms}, }, evm::OpBlockEvmFactory, - gas_limiter::AddressGasLimiter, + gas_limiter::GasLimiters, metrics::{OpRBuilderMetrics, record_flashblock_publish_timing}, primitives::reth::ExecutionInfo, runtime_ext::RuntimeExt, @@ -300,8 +300,8 @@ pub(super) struct OpPayloadBuilderInner { metrics: Arc, /// The end of builder transaction type builder_tx: BuilderTx, - /// Rate limiting based on gas. This is an optional feature. - address_gas_limiter: AddressGasLimiter, + /// Per-source gas rate limiters. `None` when the limiter is disabled. + gas_limiters: Option, /// Tokio task metrics for monitoring spawned tasks task_metrics: Arc, /// Task executor used to offload blocking work. @@ -339,7 +339,7 @@ impl OpPayloadBuilder { task_metrics: Arc, executor: Runtime, ) -> Self { - let address_gas_limiter = AddressGasLimiter::new(config.gas_limiter_config.clone()); + let gas_limiters = GasLimiters::from_args(&config.gas_limiter_config); Self { inner: Arc::new(OpPayloadBuilderInner { evm_config, @@ -351,7 +351,7 @@ impl OpPayloadBuilder { config, metrics, builder_tx, - address_gas_limiter, + gas_limiters, task_metrics, executor, }), @@ -427,7 +427,7 @@ where metrics: self.metrics.clone(), max_gas_per_txn: self.config.max_gas_per_txn, max_uncompressed_block_size: self.config.max_uncompressed_block_size, - address_gas_limiter: self.address_gas_limiter.clone(), + gas_limiters: self.gas_limiters.clone(), backrun_ctx, exclude_reverts_between_flashblocks: self.config.exclude_reverts_between_flashblocks, enable_tx_tracking_debug_logs: self.config.enable_tx_tracking_debug_logs, @@ -476,7 +476,9 @@ where ctx.enable_incremental_state_root, ); - self.address_gas_limiter.refresh(ctx.block_number()); + if let Some(limiters) = &self.gas_limiters { + limiters.refresh(ctx.block_number()); + } // Phase 1: Build the fallback block. let fallback_span = if span.is_none() { diff --git a/crates/op-rbuilder/src/builder/syncer_ctx.rs b/crates/op-rbuilder/src/builder/syncer_ctx.rs index ddc4c770c..44c106571 100644 --- a/crates/op-rbuilder/src/builder/syncer_ctx.rs +++ b/crates/op-rbuilder/src/builder/syncer_ctx.rs @@ -2,7 +2,6 @@ use crate::{ backrun_bundle::{BackrunBundleArgs, BackrunBundleGlobalPool, BackrunBundlesPayloadCtx}, builder::{BuilderConfig, OpPayloadBuilderCtx}, evm::OpBlockEvmFactory, - gas_limiter::{AddressGasLimiter, args::GasLimiterArgs}, metrics::OpRBuilderMetrics, traits::ClientBounds, }; @@ -117,7 +116,7 @@ impl OpPayloadSyncerCtx { metrics: self.metrics, max_gas_per_txn: self.max_gas_per_txn, max_uncompressed_block_size: self.max_uncompressed_block_size, - address_gas_limiter: AddressGasLimiter::new(GasLimiterArgs::default()), + gas_limiters: None, backrun_ctx, exclude_reverts_between_flashblocks: self.exclude_reverts_between_flashblocks, enable_tx_tracking_debug_logs: self.enable_tx_tracking_debug_logs, diff --git a/crates/op-rbuilder/src/gas_limiter/args.rs b/crates/op-rbuilder/src/gas_limiter/args.rs index ec7f8008b..ad058fd38 100644 --- a/crates/op-rbuilder/src/gas_limiter/args.rs +++ b/crates/op-rbuilder/src/gas_limiter/args.rs @@ -2,7 +2,7 @@ use clap::Args; #[derive(Debug, Clone, Default, PartialEq, Eq, Args)] pub struct GasLimiterArgs { - /// Enable address-based gas rate limiting + /// Enable address-based gas rate limiting for public-mempool txs. #[arg(long = "gas-limiter.enabled", env)] pub gas_limiter_enabled: bool, @@ -25,4 +25,21 @@ pub struct GasLimiterArgs { /// How many blocks to wait before cleaning up stale buckets for addresses. #[arg(long = "gas-limiter.cleanup-interval", env, default_value = "100")] pub cleanup_interval: u64, + + /// Per-address cap for the bundle limiter. Defaults to 10 million gas. + #[arg( + long = "gas-limiter.bundle-max-gas-per-address", + env, + default_value = "10000000" + )] + pub bundle_max_gas_per_address: u64, + + /// Refill rate per block for the bundle limiter. Defaults to 1 million + /// gas per block. + #[arg( + long = "gas-limiter.bundle-refill-rate-per-block", + env, + default_value = "1000000" + )] + pub bundle_refill_rate_per_block: u64, } diff --git a/crates/op-rbuilder/src/gas_limiter/mod.rs b/crates/op-rbuilder/src/gas_limiter/mod.rs index b3b60ef34..d81fde59e 100644 --- a/crates/op-rbuilder/src/gas_limiter/mod.rs +++ b/crates/op-rbuilder/src/gas_limiter/mod.rs @@ -11,12 +11,9 @@ mod metrics; #[derive(Debug, Clone)] pub struct AddressGasLimiter { - inner: Option, -} - -#[derive(Debug, Clone)] -struct AddressGasLimiterInner { - config: GasLimiterArgs, + max_gas_per_address: u64, + refill_rate_per_block: u64, + cleanup_interval: u64, // We don't need an Arc> here, we can get away with RefCell, but // the reth PayloadBuilder trait needs this to be Send + Sync address_buckets: Arc>, @@ -30,41 +27,39 @@ struct TokenBucket { } impl AddressGasLimiter { - pub fn new(config: GasLimiterArgs) -> Self { + pub fn new( + max_gas_per_address: u64, + refill_rate_per_block: u64, + cleanup_interval: u64, + ) -> Self { Self { - inner: AddressGasLimiterInner::try_new(config), + max_gas_per_address, + refill_rate_per_block, + cleanup_interval, + address_buckets: Default::default(), + metrics: Default::default(), } } /// Check if there's enough gas for this address and consume it. Returns - /// Ok(()) if there's enough otherwise returns an error. + /// `Ok(())` if there's enough, otherwise an error. pub fn consume_gas(&self, address: Address, gas_requested: u64) -> Result<(), GasLimitError> { - if let Some(inner) = &self.inner { - inner.consume_gas(address, gas_requested) - } else { - Ok(()) - } - } + let start = Instant::now(); + let result = self.consume_gas_inner(address, gas_requested); - /// Should be called upon each new block. Refills buckets/Garbage collection - pub fn refresh(&self, block_number: u64) { - if let Some(inner) = self.inner.as_ref() { - inner.refresh(block_number) - } + self.metrics.record_gas_check(&result, start.elapsed()); + + result.map(|_| ()) } -} -impl AddressGasLimiterInner { - fn try_new(config: GasLimiterArgs) -> Option { - if !config.gas_limiter_enabled { - return None; - } + /// Should be called upon each new block. Refills buckets and runs + /// periodic garbage collection. + pub fn refresh(&self, block_number: u64) { + let start = Instant::now(); + let removed_addresses = self.refresh_inner(block_number); - Some(Self { - config, - address_buckets: Default::default(), - metrics: Default::default(), - }) + self.metrics + .record_refresh(removed_addresses, start.elapsed()); } fn consume_gas_inner( @@ -73,14 +68,10 @@ impl AddressGasLimiterInner { gas_requested: u64, ) -> Result { let mut created_new_bucket = false; - let mut bucket = self - .address_buckets - .entry(address) - // if we don't find a bucket we need to initialize a new one - .or_insert_with(|| { - created_new_bucket = true; - TokenBucket::new(self.config.max_gas_per_address) - }); + let mut bucket = self.address_buckets.entry(address).or_insert_with(|| { + created_new_bucket = true; + TokenBucket::new(self.max_gas_per_address) + }); if gas_requested > bucket.available { return Err(GasLimitError::AddressLimitExceeded { @@ -95,41 +86,24 @@ impl AddressGasLimiterInner { Ok(created_new_bucket) } - fn consume_gas(&self, address: Address, gas_requested: u64) -> Result<(), GasLimitError> { - let start = Instant::now(); - let result = self.consume_gas_inner(address, gas_requested); - - self.metrics.record_gas_check(&result, start.elapsed()); - - result.map(|_| ()) - } - fn refresh_inner(&self, block_number: u64) -> usize { let active_addresses = self.address_buckets.len(); self.address_buckets.iter_mut().for_each(|mut bucket| { bucket.available = min( bucket.capacity, - bucket.available + self.config.refill_rate_per_block, + bucket.available + self.refill_rate_per_block, ) }); // Only clean up stale buckets every `cleanup_interval` blocks - if block_number.is_multiple_of(self.config.cleanup_interval) { + if block_number.is_multiple_of(self.cleanup_interval) { self.address_buckets .retain(|_, bucket| bucket.available < bucket.capacity); } active_addresses - self.address_buckets.len() } - - fn refresh(&self, block_number: u64) { - let start = Instant::now(); - let removed_addresses = self.refresh_inner(block_number); - - self.metrics - .record_refresh(removed_addresses, start.elapsed()); - } } impl TokenBucket { @@ -141,17 +115,60 @@ impl TokenBucket { } } +/// Holds one [`AddressGasLimiter`] per tx source. The two pools are always +/// configured together — callers represent the disabled state with +/// `Option` rather than per-field optionality. +#[derive(Debug, Clone)] +pub struct GasLimiters { + /// Limiter applied to public-mempool txs. + pub mempool: AddressGasLimiter, + /// Limiter applied to bundle txs (including backruns). + pub bundle: AddressGasLimiter, +} + +impl GasLimiters { + /// Build per-source limiters from the flat `GasLimiterArgs`. Returns + /// `None` when the limiter is disabled. When enabled, both pools are + /// constructed with their respective capacity/refill settings. + pub fn from_args(args: &GasLimiterArgs) -> Option { + if !args.gas_limiter_enabled { + return None; + } + + Some(Self { + mempool: AddressGasLimiter::new( + args.max_gas_per_address, + args.refill_rate_per_block, + args.cleanup_interval, + ), + bundle: AddressGasLimiter::new( + args.bundle_max_gas_per_address, + args.bundle_refill_rate_per_block, + args.cleanup_interval, + ), + }) + } + + /// Refresh both limiters. Call once per new block. + pub fn refresh(&self, block_number: u64) { + self.mempool.refresh(block_number); + self.bundle.refresh(block_number); + } +} + #[cfg(test)] mod tests { use super::*; use alloy_primitives::Address; - fn create_test_config(max_gas: u64, refill_rate: u64, cleanup_interval: u64) -> GasLimiterArgs { + fn enabled_args(max_gas: u64, refill_rate: u64, cleanup_interval: u64) -> GasLimiterArgs { GasLimiterArgs { gas_limiter_enabled: true, max_gas_per_address: max_gas, refill_rate_per_block: refill_rate, cleanup_interval, + bundle_max_gas_per_address: max_gas, + bundle_refill_rate_per_block: refill_rate, } } @@ -161,8 +178,7 @@ mod tests { #[test] fn test_basic_refill() { - let config = create_test_config(1000, 200, 10); - let limiter = AddressGasLimiter::new(config); + let limiter = AddressGasLimiter::new(1000, 200, 10); // Consume all gas assert!(limiter.consume_gas(test_address(), 1000).is_ok()); @@ -176,8 +192,7 @@ mod tests { #[test] fn test_over_capacity_request() { - let config = create_test_config(1000, 100, 10); - let limiter = AddressGasLimiter::new(config); + let limiter = AddressGasLimiter::new(1000, 100, 10); // Request more than capacity should fail let result = limiter.consume_gas(test_address(), 1500); @@ -194,8 +209,7 @@ mod tests { #[test] fn test_multiple_users() { // Simulate more realistic scenario - let config = create_test_config(10_000_000, 1_000_000, 100); // 10M max, 1M refill - let limiter = AddressGasLimiter::new(config); + let limiter = AddressGasLimiter::new(10_000_000, 1_000_000, 100); // 10M max, 1M refill let searcher1 = Address::from([0x1; 20]); let searcher2 = Address::from([0x2; 20]); @@ -224,8 +238,7 @@ mod tests { #[test] fn test_bucket_cleanup() { // Test that unused buckets get cleaned up properly - let config = create_test_config(1000, 1000, 10); - let limiter = AddressGasLimiter::new(config); + let limiter = AddressGasLimiter::new(1000, 1000, 10); let addr1 = Address::from([0x1; 20]); let addr2 = Address::from([0x2; 20]); @@ -234,8 +247,7 @@ mod tests { assert!(limiter.consume_gas(addr1, 100).is_ok()); assert!(limiter.consume_gas(addr2, 100).is_ok()); - let inner = limiter.inner.as_ref().unwrap(); - assert_eq!(inner.address_buckets.len(), 2); + assert_eq!(limiter.address_buckets.len(), 2); // Refill for several blocks - addr1 stays at full capacity (unused) // but addr2 continues to be used @@ -249,14 +261,48 @@ mod tests { } } - // After cleanup at block 10 (multiple of 5), addr1 should be removed - // because it's at full capacity (unused), while addr2 remains + // After cleanup at block 10 (multiple of cleanup_interval), addr1 + // should be removed because it's at full capacity (unused), while + // addr2 remains. assert_eq!( - inner.address_buckets.len(), + limiter.address_buckets.len(), 1, "Unused bucket (addr1) should have been cleaned up" ); - assert!(inner.address_buckets.contains_key(&addr2)); - assert!(!inner.address_buckets.contains_key(&addr1)); + assert!(limiter.address_buckets.contains_key(&addr2)); + assert!(!limiter.address_buckets.contains_key(&addr1)); + } + + #[test] + fn test_gas_limiters_separate_bundle_and_mempool_pools() { + // Tight bundle limit, generous mempool limit. + let args = GasLimiterArgs { + bundle_max_gas_per_address: 500, + bundle_refill_rate_per_block: 100, + ..enabled_args(10_000_000, 1_000_000, 10) + }; + let limiters = GasLimiters::from_args(&args).expect("limiter enabled"); + let addr = test_address(); + + // Bundle pool: 500 capacity. Drain it. + assert!(limiters.bundle.consume_gas(addr, 500).is_ok()); + assert!(limiters.bundle.consume_gas(addr, 1).is_err()); + + // Mempool pool: 10M capacity, untouched by the bundle activity. + assert!(limiters.mempool.consume_gas(addr, 5_000_000).is_ok()); + assert!(limiters.mempool.consume_gas(addr, 5_000_000).is_ok()); + assert!(limiters.mempool.consume_gas(addr, 1).is_err()); + + // Refresh both pools at once. + limiters.refresh(1); + assert!(limiters.bundle.consume_gas(addr, 100).is_ok()); + assert!(limiters.bundle.consume_gas(addr, 1).is_err()); + assert!(limiters.mempool.consume_gas(addr, 1_000_000).is_ok()); + } + + #[test] + fn test_gas_limiters_disabled() { + let args = GasLimiterArgs::default(); // gas_limiter_enabled = false + assert!(GasLimiters::from_args(&args).is_none()); } } diff --git a/crates/op-rbuilder/src/metrics.rs b/crates/op-rbuilder/src/metrics.rs index 20c692a02..3b558640a 100644 --- a/crates/op-rbuilder/src/metrics.rs +++ b/crates/op-rbuilder/src/metrics.rs @@ -139,8 +139,6 @@ pub struct OpRBuilderMetrics { pub reverted_tx_gas_used: Histogram, /// Gas used by reverted transactions in the latest block pub payload_reverted_tx_gas_used: Gauge, - /// Histogram of tx simulation duration - pub tx_simulation_duration: Histogram, /// Byte size of transactions pub tx_byte_size: Histogram, /// How much less flashblocks we issue to be on time with block construction @@ -264,6 +262,29 @@ pub fn record_flashblock_publish_timing(flashblock_index: u64, offset_ms: f64) { .record(offset_ms); } +/// Record tx simulation duration with labels for tx source (`bundle` vs. +/// `mempool`) and EVM outcome (`success` / `reverted` / `halted`). Sum +/// across labels in PromQL to recover the prior unlabeled total. +pub fn record_tx_simulation_duration( + duration: std::time::Duration, + is_bundle: bool, + result: &revm::context::result::ExecutionResult, +) { + use revm::context::result::ExecutionResult; + let kind = if is_bundle { "bundle" } else { "mempool" }; + let outcome = match result { + ExecutionResult::Success { .. } => "success", + ExecutionResult::Revert { .. } => "reverted", + ExecutionResult::Halt { .. } => "halted", + }; + histogram!( + "op_rbuilder_tx_simulation_duration", + "kind" => kind, + "result" => outcome, + ) + .record(duration); +} + /// Set gauge metrics for some flags so we can inspect which ones are set /// and which ones aren't. pub fn record_flag_gauge_metrics(builder_args: &OpRbuilderArgs) { diff --git a/crates/op-rbuilder/src/tests/gas_limiter.rs b/crates/op-rbuilder/src/tests/gas_limiter.rs index d44af1789..1c50a8eda 100644 --- a/crates/op-rbuilder/src/tests/gas_limiter.rs +++ b/crates/op-rbuilder/src/tests/gas_limiter.rs @@ -16,6 +16,7 @@ use tracing::info; max_gas_per_address: 200000, // 200k gas per address - low for testing refill_rate_per_block: 100000, // 100k gas refill per block cleanup_interval: 100, + ..Default::default() }, ..Default::default() })]