diff --git a/crates/op-rbuilder/src/builder/candidate.rs b/crates/op-rbuilder/src/builder/candidate.rs new file mode 100644 index 00000000..1c7a3733 --- /dev/null +++ b/crates/op-rbuilder/src/builder/candidate.rs @@ -0,0 +1,601 @@ +use crate::{ + builder::{ + best_txs::{FlashblockPoolTxCursor, FlashblockTxTracker}, + builder_tx::{BuilderTransactions, reserve_builder_tx_budget}, + context::OpPayloadJobCtx, + payload::{FlashblocksState, OpPayloadBuilder}, + state_root::StateRootCalculator, + }, + primitives::reth::ExecutionInfo, + traits::{ClientBounds, PoolBounds}, +}; +use eyre::WrapErr as _; +use op_alloy_rpc_types_engine::OpFlashblockPayload; +use reth_optimism_node::OpBuiltPayload; +use reth_payload_util::BestPayloadTransactions; +use reth_provider::{ + HashedPostStateProvider, ProviderError, StateRootProvider, StorageRootProvider, +}; +use reth_revm::{ + State, + db::{CacheState, TransitionState}, +}; +use revm::Database; +use std::{ + mem, + time::{Duration, Instant}, +}; +use tokio_util::sync::CancellationToken; +use tracing::error; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum CandidateKind { + Empty, + PoolBacked, +} + +/// Preserves the distinct cancellation contracts of the naive and continuous builders. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum InterruptPolicy { + JobOnly, + JobAndFlashblock, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum CandidateCheckpoint { + AfterExecution, + BeforeAssembly, + AfterAssembly, +} + +#[derive(Debug, Clone, Copy)] +pub(crate) struct CandidateBudgetTargets { + gas: u64, + da: Option, + da_footprint: Option, + max_uncompressed_block_size: Option, +} + +impl CandidateBudgetTargets { + pub(crate) fn from_fb_state(fb_state: &FlashblocksState) -> Self { + Self { + gas: fb_state.target_gas_for_batch(), + da: fb_state.target_da_for_batch(), + da_footprint: fb_state.target_da_footprint_for_batch(), + max_uncompressed_block_size: None, + } + } +} + +#[derive(Debug, Default, Clone, Copy)] +pub(crate) struct CandidateTimings { + pub(crate) pool_fetch: Option, + pub(crate) execution: Option, + pub(crate) assemble: Duration, + pub(crate) total: Duration, +} + +pub(crate) struct CarriedCandidate { + pub(crate) cache: CacheState, + pub(crate) transition: Option, + pub(crate) info: ExecutionInfo, + pub(crate) fb_state: FlashblocksState, + pub(crate) tx_tracker: FlashblockTxTracker, + pub(crate) state_root_calc: StateRootCalculator, +} + +impl CarriedCandidate { + pub(crate) fn restore( + self, + state: &mut State, + info: &mut ExecutionInfo, + fb_state: &mut FlashblocksState, + tx_tracker: &mut FlashblockTxTracker, + state_root_calc: &mut StateRootCalculator, + ) { + state.cache = self.cache; + state.transition_state = self.transition; + *info = self.info; + *fb_state = self.fb_state; + *tx_tracker = self.tx_tracker; + *state_root_calc = self.state_root_calc; + } +} + +pub(crate) struct BuiltCandidate { + pub(crate) new_payload: OpBuiltPayload, + pub(crate) fb_payload: OpFlashblockPayload, + pub(crate) next_fb_state: FlashblocksState, + pub(crate) carried: CarriedCandidate, + pub(crate) timings: CandidateTimings, +} + +pub(crate) struct CancelledCandidate { + pub(crate) carried: CarriedCandidate, + pub(crate) timings: CandidateTimings, +} + +#[allow(clippy::large_enum_variant)] +pub(crate) enum CandidateOutcome { + Built(BuiltCandidate), + Cancelled(CancelledCandidate), + SealFailed(CandidateSealFailure), +} + +pub(crate) enum CandidateSealFailure { + Input(eyre::Report), + Assemble(eyre::Report), +} + +impl CandidateSealFailure { + pub(crate) fn into_continuous_error(self) -> eyre::Report { + match self { + Self::Input(err) => err.wrap_err("failed to construct block assembly input"), + Self::Assemble(err) => err.wrap_err("failed to assemble candidate"), + } + } +} + +impl OpPayloadBuilder +where + Pool: PoolBounds + 'static, + Client: ClientBounds + 'static, + BuilderTx: BuilderTransactions + Send + Sync + 'static, +{ + pub(crate) fn prepare_candidate_budget< + DB: Database + std::fmt::Debug + AsRef

, + P: StateRootProvider + HashedPostStateProvider + StorageRootProvider, + >( + &self, + ctx: &OpPayloadJobCtx, + state_provider: impl reth::providers::StateProvider + Clone, + state: &mut State, + info: &mut ExecutionInfo, + fb_state: &FlashblocksState, + ) -> CandidateBudgetTargets { + let mut targets = CandidateBudgetTargets::from_fb_state(fb_state); + + let builder_txs = self + .builder_tx() + .add_builder_txs( + &state_provider, + info, + &ctx.builder_tx_env(), + state, + true, + fb_state.is_first_flashblock(), + fb_state.is_last_flashblock(), + ) + .inspect_err( + |err| error!(target: "payload_builder", %err, "Error simulating builder txs"), + ) + .unwrap_or_default(); + + targets.max_uncompressed_block_size = reserve_builder_tx_budget( + &builder_txs, + &mut targets.gas, + &mut targets.da, + &mut targets.da_footprint, + info.da_footprint_scalar, + ctx.max_uncompressed_block_size, + info.cumulative_uncompressed_bytes, + ); + targets + } + + #[expect(clippy::too_many_arguments)] + pub(crate) fn build_candidate< + DB: Database + std::fmt::Debug + AsRef

, + P: StateRootProvider + HashedPostStateProvider + StorageRootProvider, + >( + &self, + kind: CandidateKind, + budget_targets: CandidateBudgetTargets, + ctx: &OpPayloadJobCtx, + state_provider: impl reth::providers::StateProvider + Clone, + state: &mut State, + mut info: ExecutionInfo, + mut fb_state: FlashblocksState, + mut tx_tracker: FlashblockTxTracker, + mut state_root_calc: StateRootCalculator, + cancellation: &CancellationToken, + interrupt_policy: InterruptPolicy, + ) -> eyre::Result { + let total_start = Instant::now(); + let flashblock_index = fb_state.flashblock_index(); + let target_gas_for_batch = budget_targets.gas; + let target_da_for_batch = budget_targets.da; + let target_da_footprint_for_batch = budget_targets.da_footprint; + + let mut pool_fetch = None; + let mut execution = None; + if kind.uses_pool() { + let pool_fetch_start = Instant::now(); + let mut best_txs = FlashblockPoolTxCursor::new(&mut tx_tracker); + best_txs.refresh_iterator( + BestPayloadTransactions::new( + self.pool() + .best_transactions_with_attributes(ctx.best_transaction_attributes()), + ), + flashblock_index, + ); + pool_fetch = Some(pool_fetch_start.elapsed()); + + let execution_start = Instant::now(); + let execution_cancelled = ctx + .execute_best_transactions( + &mut info, + state, + &mut best_txs, + target_gas_for_batch.min(ctx.block_gas_limit()), + target_da_for_batch, + target_da_footprint_for_batch, + budget_targets.max_uncompressed_block_size, + flashblock_index, + ) + .wrap_err("failed to execute best transactions")? + .is_some(); + + let new_transactions = fb_state + .slice_new_transactions(&info.executed_transactions) + .iter() + .map(|tx| tx.tx_hash()) + .collect::>(); + best_txs.mark_committed(new_transactions); + drop(best_txs); + + if !info.reverted_bundle_tx_hashes.is_empty() { + self.pool() + .remove_transactions(mem::take(&mut info.reverted_bundle_tx_hashes)); + } + if interrupt_policy.should_interrupt( + kind, + CandidateCheckpoint::AfterExecution, + cancellation.is_cancelled(), + ctx.cancel().is_cancelled(), + execution_cancelled, + ) { + return Ok(CandidateOutcome::Cancelled(CancelledCandidate { + carried: take_carried(state, info, fb_state, tx_tracker, state_root_calc), + timings: CandidateTimings { + pool_fetch, + total: total_start.elapsed(), + ..Default::default() + }, + })); + } + execution = Some(execution_start.elapsed()); + } + + if let Err(err) = self.builder_tx().add_builder_txs( + &state_provider, + &mut info, + &ctx.builder_tx_env(), + state, + false, + fb_state.is_first_flashblock(), + fb_state.is_last_flashblock(), + ) { + error!(target: "payload_builder", %err, "Error adding bottom builder txs"); + } + + if interrupt_policy.should_interrupt( + kind, + CandidateCheckpoint::BeforeAssembly, + cancellation.is_cancelled(), + ctx.cancel().is_cancelled(), + false, + ) { + return Ok(cancellation_before_assemble( + state, + info, + fb_state, + tx_tracker, + state_root_calc, + CandidateTimings { + pool_fetch, + execution, + total: total_start.elapsed(), + ..Default::default() + }, + )); + } + + let assemble_start = Instant::now(); + let input = match ctx.block_assembly_input() { + Ok(input) => input, + Err(err) => { + return Ok(CandidateOutcome::SealFailed(CandidateSealFailure::Input( + err.into(), + ))); + } + }; + let build_result = input.assemble( + state, + Some(&mut fb_state), + &mut info, + &mut state_root_calc, + ctx.metrics.clone(), + ctx.enable_tx_tracking_debug_logs, + ); + let assemble = assemble_start.elapsed(); + let (new_payload, mut fb_payload) = match build_result { + Ok(output) => output, + Err(err) => { + return Ok(CandidateOutcome::SealFailed( + CandidateSealFailure::Assemble(err.into()), + )); + } + }; + + fb_payload.index = flashblock_index; + fb_payload.base = None; + + debug_assert!(!interrupt_policy.should_interrupt( + kind, + CandidateCheckpoint::AfterAssembly, + cancellation.is_cancelled(), + ctx.cancel().is_cancelled(), + false, + )); + + let next_fb_state = + fb_state.next_after_seal(target_da_for_batch, target_da_footprint_for_batch); + let carried = take_carried(state, info, fb_state, tx_tracker, state_root_calc); + Ok(completed_candidate(BuiltCandidate { + new_payload, + fb_payload, + next_fb_state, + carried, + timings: CandidateTimings { + pool_fetch, + execution, + assemble, + total: total_start.elapsed(), + }, + })) + } +} + +impl CandidateKind { + fn uses_pool(self) -> bool { + self == Self::PoolBacked + } +} + +impl InterruptPolicy { + fn should_interrupt( + self, + kind: CandidateKind, + checkpoint: CandidateCheckpoint, + job_cancelled: bool, + flashblock_cancelled: bool, + execution_cancelled: bool, + ) -> bool { + if kind == CandidateKind::Empty || checkpoint == CandidateCheckpoint::AfterAssembly { + return false; + } + match self { + Self::JobOnly => checkpoint == CandidateCheckpoint::AfterExecution && job_cancelled, + Self::JobAndFlashblock => { + job_cancelled + || flashblock_cancelled + || (checkpoint == CandidateCheckpoint::AfterExecution && execution_cancelled) + } + } + } +} + +fn cancellation_before_assemble( + state: &mut State, + info: ExecutionInfo, + fb_state: FlashblocksState, + tx_tracker: FlashblockTxTracker, + state_root_calc: StateRootCalculator, + timings: CandidateTimings, +) -> CandidateOutcome { + CandidateOutcome::Cancelled(CancelledCandidate { + carried: take_carried(state, info, fb_state, tx_tracker, state_root_calc), + timings, + }) +} + +fn take_carried( + state: &mut State, + info: ExecutionInfo, + fb_state: FlashblocksState, + tx_tracker: FlashblockTxTracker, + state_root_calc: StateRootCalculator, +) -> CarriedCandidate { + CarriedCandidate { + cache: mem::take(&mut state.cache), + transition: state.transition_state.take(), + info, + fb_state, + tx_tracker, + state_root_calc, + } +} + +fn completed_candidate(candidate: BuiltCandidate) -> CandidateOutcome { + CandidateOutcome::Built(candidate) +} + +#[cfg(test)] +mod tests { + use super::*; + use alloy_consensus::{BlockBody, Header}; + use alloy_primitives::U256; + use reth_optimism_primitives::OpBlock; + use reth_payload_builder::PayloadId; + use reth_primitives_traits::Block as _; + use revm::database::EmptyDB; + use std::sync::Arc; + + fn budget_state() -> FlashblocksState { + FlashblocksState::new(5).with_batch_limits( + 100_000, + Some(1_000), + Some(500), + 100_000, + Some(1_000), + Some(500), + ) + } + + #[test] + fn empty_candidate_skips_only_pool_segment() { + assert!(!CandidateKind::Empty.uses_pool()); + assert!(CandidateKind::PoolBacked.uses_pool()); + } + + #[test] + fn budget_carry_advances_across_two_candidates() { + let first = budget_state().next_after_seal(Some(700), Some(400)); + assert_eq!(first.flashblock_index(), 1); + assert_eq!(first.target_gas_for_batch(), 200_000); + assert_eq!(first.target_da_for_batch(), Some(1_700)); + assert_eq!(first.target_da_footprint_for_batch(), Some(900)); + + let second = first.next_after_seal(Some(1_200), Some(650)); + assert_eq!(second.flashblock_index(), 2); + assert_eq!(second.target_gas_for_batch(), 300_000); + assert_eq!(second.target_da_for_batch(), Some(2_200)); + assert_eq!(second.target_da_footprint_for_batch(), Some(1_150)); + } + + #[test] + fn cancellation_before_assemble_returns_carried_state_without_output() { + let mut state = State::builder() + .with_database(EmptyDB::default()) + .with_bundle_update() + .build(); + let fb_state = budget_state(); + let outcome = cancellation_before_assemble( + &mut state, + ExecutionInfo::default(), + fb_state, + FlashblockTxTracker::default(), + StateRootCalculator::new(true, false), + CandidateTimings::default(), + ); + + let CandidateOutcome::Cancelled(cancelled) = outcome else { + panic!("cancellation must not publish candidate outputs"); + }; + assert_eq!(cancelled.carried.fb_state.flashblock_index(), 0); + assert!(state.transition_state.is_none()); + } + + #[test] + fn job_only_ignores_flashblock_and_execution_cancellation() { + let job = CancellationToken::new(); + let flashblock = CancellationToken::new(); + flashblock.cancel(); + + assert!(!InterruptPolicy::JobOnly.should_interrupt( + CandidateKind::PoolBacked, + CandidateCheckpoint::AfterExecution, + job.is_cancelled(), + flashblock.is_cancelled(), + true, + )); + } + + #[test] + fn empty_candidate_ignores_all_cancellation() { + let job = CancellationToken::new(); + let flashblock = CancellationToken::new(); + job.cancel(); + flashblock.cancel(); + + assert!(!InterruptPolicy::JobAndFlashblock.should_interrupt( + CandidateKind::Empty, + CandidateCheckpoint::AfterExecution, + job.is_cancelled(), + flashblock.is_cancelled(), + true, + )); + } + + #[test] + fn cancelled_candidate_carried_state_is_restorable() { + let mut state = State::builder() + .with_database(EmptyDB::default()) + .with_bundle_update() + .build(); + state.transition_state = Some(TransitionState::default()); + let mut restored_info = ExecutionInfo::default(); + let mut restored_fb_state = FlashblocksState::default(); + let mut restored_tracker = FlashblockTxTracker::default(); + let mut restored_root_calc = StateRootCalculator::new(false, false); + let mut carried_info = ExecutionInfo::default(); + let expected_fees = alloy_primitives::U256::from_limbs([42, 0, 0, 0]); + carried_info.total_fees = expected_fees; + let carried_fb_state = budget_state().next_after_seal(Some(700), Some(400)); + let cancelled = cancellation_before_assemble( + &mut state, + carried_info, + carried_fb_state, + FlashblockTxTracker::default(), + StateRootCalculator::new(true, false), + CandidateTimings::default(), + ); + let CandidateOutcome::Cancelled(cancelled) = cancelled else { + panic!("expected cancelled candidate"); + }; + + cancelled.carried.restore( + &mut state, + &mut restored_info, + &mut restored_fb_state, + &mut restored_tracker, + &mut restored_root_calc, + ); + + assert!(state.transition_state.is_some()); + assert_eq!(restored_info.total_fees, expected_fees); + assert_eq!(restored_fb_state.flashblock_index(), 1); + } + + #[test] + fn assembled_candidate_is_not_discarded_by_primitive_policy() { + let job = CancellationToken::new(); + let flashblock = CancellationToken::new(); + job.cancel(); + flashblock.cancel(); + + // The last primitive checkpoint is before assembly; callers decide + // whether a successfully assembled candidate may be published. + assert!(!InterruptPolicy::JobAndFlashblock.should_interrupt( + CandidateKind::PoolBacked, + CandidateCheckpoint::AfterAssembly, + job.is_cancelled(), + flashblock.is_cancelled(), + false, + )); + + let carried = CarriedCandidate { + cache: CacheState::default(), + transition: None, + info: ExecutionInfo::default(), + fb_state: budget_state(), + tx_tracker: FlashblockTxTracker::default(), + state_root_calc: StateRootCalculator::new(true, false), + }; + let outcome = completed_candidate(BuiltCandidate { + new_payload: OpBuiltPayload::new( + PayloadId::new([0; 8]), + Arc::new(OpBlock::new(Header::default(), BlockBody::default()).seal_slow()), + U256::ZERO, + None, + ), + fb_payload: OpFlashblockPayload::default(), + next_fb_state: budget_state(), + carried, + timings: CandidateTimings::default(), + }); + + assert!(matches!(outcome, CandidateOutcome::Built(_))); + } +} diff --git a/crates/op-rbuilder/src/builder/continuous/candidate_loop.rs b/crates/op-rbuilder/src/builder/continuous/candidate_loop.rs index 248b0c16..17b7463c 100644 --- a/crates/op-rbuilder/src/builder/continuous/candidate_loop.rs +++ b/crates/op-rbuilder/src/builder/continuous/candidate_loop.rs @@ -4,8 +4,9 @@ use super::{ }; use crate::{ builder::{ - best_txs::{FlashblockPoolTxCursor, FlashblockTxTracker}, - builder_tx::{BuilderTransactions, reserve_builder_tx_budget}, + best_txs::FlashblockTxTracker, + builder_tx::BuilderTransactions, + candidate::{BuiltCandidate, CandidateKind, CandidateOutcome, InterruptPolicy}, context::OpPayloadJobCtx, payload::{FlashblocksState, OpPayloadBuilder}, state_root::StateRootCalculator, @@ -14,22 +15,14 @@ use crate::{ traits::{ClientBounds, PoolBounds}, }; use alloy_primitives::U256; -use eyre::WrapErr as _; -use op_alloy_rpc_types_engine::OpFlashblockPayload; -use reth_optimism_node::OpBuiltPayload; -use reth_payload_util::BestPayloadTransactions; use reth_provider::{ HashedPostStateProvider, ProviderError, StateRootProvider, StorageRootProvider, }; use reth_revm::State; use revm::Database; -use std::{ - mem, - sync::atomic::Ordering, - time::{Duration, Instant}, -}; +use std::{sync::atomic::Ordering, time::Duration}; use tokio_util::sync::CancellationToken; -use tracing::{error, field, info, metadata::Level, span, warn}; +use tracing::{field, info, metadata::Level, span, warn}; // === Build / candidate loop internals ======================= // @@ -43,60 +36,6 @@ where Client: ClientBounds + 'static, BuilderTx: BuilderTransactions + Send + Sync + 'static, { - #[expect(clippy::too_many_arguments)] - fn build_empty_flashblock_candidate< - DB: Database + std::fmt::Debug + AsRef

, - P: StateRootProvider + HashedPostStateProvider + StorageRootProvider, - >( - &self, - ctx: &OpPayloadJobCtx, - fb_state: &mut FlashblocksState, - info: &mut ExecutionInfo, - state: &mut State, - state_provider: impl reth::providers::StateProvider + Clone, - state_root_calc: &mut StateRootCalculator, - target_da_for_batch: Option, - target_da_footprint_for_batch: Option, - ) -> eyre::Result<(FlashblocksState, OpBuiltPayload, OpFlashblockPayload)> { - if let Err(e) = self.builder_tx().add_builder_txs( - &state_provider, - info, - &ctx.builder_tx_env(), - state, - false, - fb_state.is_first_flashblock(), - fb_state.is_last_flashblock(), - ) { - error!( - target: "payload_builder", - "Error adding bottom builder txs to empty flashblock candidate: {}", - e - ); - } - - let (new_payload, mut fb_payload) = ctx - .block_assembly_input() - .wrap_err("failed to construct block assembly input")? - .assemble( - state, - Some(fb_state), - info, - state_root_calc, - ctx.metrics.clone(), - ctx.enable_tx_tracking_debug_logs, - ) - .wrap_err("failed to build empty flashblock candidate")?; - - fb_payload.index = fb_state.flashblock_index(); - fb_payload.base = None; - - Ok(( - fb_state.next_after_seal(target_da_for_batch, target_da_footprint_for_batch), - new_payload, - fb_payload, - )) - } - #[expect(clippy::too_many_arguments)] pub(super) fn build_continuous_flashblock< DB: Database + std::fmt::Debug + AsRef

, @@ -114,50 +53,23 @@ where shared_best: &SharedBest, ) -> eyre::Result { let flashblock_index = fb_state.flashblock_index(); - let mut target_gas_for_batch = fb_state.target_gas_for_batch(); - let mut target_da_for_batch = fb_state.target_da_for_batch(); - let mut target_da_footprint_for_batch = fb_state.target_da_footprint_for_batch(); - info!( target: "payload_builder", block_number = ctx.block_number(), flashblock_index, - target_gas = target_gas_for_batch, + target_gas = fb_state.target_gas_for_batch(), gas_used = info.cumulative_gas_used, - target_da = target_da_for_batch, + target_da = fb_state.target_da_for_batch(), da_used = info.cumulative_da_bytes_used, block_gas_used = ctx.block_gas_limit(), - target_da_footprint = target_da_footprint_for_batch, + target_da_footprint = fb_state.target_da_footprint_for_batch(), "continuous: starting candidate loop", ); - let builder_txs = self - .builder_tx() - .add_builder_txs( - &state_provider, - info, - &ctx.builder_tx_env(), - state, - true, - fb_state.is_first_flashblock(), - fb_state.is_last_flashblock(), - ) - .inspect_err( - |e| error!(target: "payload_builder", "Error simulating builder txs: {}", e), - ) - .unwrap_or_default(); - - let max_uncompressed_block_size = reserve_builder_tx_budget( - &builder_txs, - &mut target_gas_for_batch, - &mut target_da_for_batch, - &mut target_da_footprint_for_batch, - info.da_footprint_scalar, - ctx.max_uncompressed_block_size, - info.cumulative_uncompressed_bytes, - ); + let budget_targets = + self.prepare_candidate_budget(ctx, &state_provider, state, info, fb_state); - // Each candidate must start with the same base state. + // Each candidate must start with the same prepared base state. let base_cache = state.cache.clone(); let base_transition = state.transition_state.clone(); let base_info = info.clone(); @@ -181,21 +93,23 @@ where ctx.address_limiter() .restore_pending(&base_limiter_snapshot); - let mut empty_candidate_info = base_info.clone(); - let mut empty_candidate_fb_state = base_fb_state.clone(); - let mut empty_candidate_state_root_calc = base_state_root_calc.clone(); - let empty_build_start = Instant::now(); - let empty_candidate = self.build_empty_flashblock_candidate( - ctx, - &mut empty_candidate_fb_state, - &mut empty_candidate_info, - state, - &state_provider, - &mut empty_candidate_state_root_calc, - target_da_for_batch, - target_da_footprint_for_batch, - ); - let empty_build_duration = empty_build_start.elapsed(); + state.cache = base_cache.clone(); + state.transition_state = base_transition.clone(); + let empty_candidate = self + .build_candidate( + CandidateKind::Empty, + budget_targets, + ctx, + &state_provider, + state, + base_info.clone(), + base_fb_state.clone(), + base_tx_tracker.clone(), + base_state_root_calc.clone(), + block_cancel, + InterruptPolicy::JobAndFlashblock, + ) + .expect("empty candidates report sealing failures as outcomes"); candidates_evaluated += 1; // Baseline fees for the fee-improvement metric: whatever the first @@ -204,17 +118,23 @@ where let mut first_candidate_fees: Option = None; match empty_candidate { - Ok((next_flashblock_state, new_payload, fb_payload)) => { - let empty_fees = empty_candidate_info.total_fees; + CandidateOutcome::Built(BuiltCandidate { + new_payload, + fb_payload, + next_fb_state, + carried, + timings, + }) => { + let empty_fees = carried.info.total_fees; let candidate = BestCandidate { total_fees: empty_fees, - cache: state.cache.clone(), - transition: state.transition_state.clone(), - info: empty_candidate_info, - fb_state: empty_candidate_fb_state, - tx_tracker: base_tx_tracker.clone(), - result: (next_flashblock_state, new_payload, fb_payload), - build_duration: empty_build_duration, + cache: carried.cache, + transition: carried.transition, + info: carried.info, + fb_state: carried.fb_state, + tx_tracker: carried.tx_tracker, + result: (next_fb_state, new_payload, fb_payload), + build_duration: timings.total, // no pool fetch transaction_pool_fetch_duration: None, // not measured for empty candidate @@ -222,15 +142,21 @@ where limiter_snapshot: ctx.address_limiter().snapshot_pending(), candidates_evaluated, candidates_improved: candidates_improved + 1, - state_root_calc: empty_candidate_state_root_calc, + state_root_calc: carried.state_root_calc, }; shared_best.store(candidate.clone()); best = Some(candidate); first_candidate_fees = Some(empty_fees); candidates_improved += 1; } - Err(err) => { + CandidateOutcome::Cancelled(cancelled) => { + cancelled + .carried + .restore(state, info, fb_state, tx_tracker, state_root_calc); + } + CandidateOutcome::SealFailed(failure) => { ctx.metrics.invalid_built_blocks_count.increment(1); + let err = failure.into_continuous_error(); warn!( target: "payload_builder", ?err, @@ -268,135 +194,64 @@ where ); let _candidate_guard = candidate_span.enter(); - // Per-candidate build timing so a winning candidate can carry its - // own single-build duration. This is what gets recorded on - // `flashblock_build_duration` at publish time, keeping that - // metric comparable with the non-continuous path. - let candidate_build_start = Instant::now(); - + // Include clone/reset work to stay comparable with the naive builder. + let candidate_build_start = std::time::Instant::now(); state.cache = base_cache.clone(); state.transition_state = base_transition.clone(); ctx.address_limiter() .restore_pending(&base_limiter_snapshot); - let mut sim_info = base_info.clone(); - let mut sim_fb_state = base_fb_state.clone(); - let mut sim_tx_tracker = base_tx_tracker.clone(); - let mut sim_state_root_calc = base_state_root_calc.clone(); - - let mut best_txs = FlashblockPoolTxCursor::new(&mut sim_tx_tracker); - let best_txs_start_time = Instant::now(); - best_txs.refresh_iterator( - BestPayloadTransactions::new( - self.pool() - .best_transactions_with_attributes(ctx.best_transaction_attributes()), - ), - flashblock_index, - ); - let transaction_pool_fetch_time = best_txs_start_time.elapsed(); - - let exec_cancelled = ctx - .execute_best_transactions( - &mut sim_info, - state, - &mut best_txs, - target_gas_for_batch.min(ctx.block_gas_limit()), - target_da_for_batch, - target_da_footprint_for_batch, - max_uncompressed_block_size, - sim_fb_state.flashblock_index(), - ) - .wrap_err("failed to execute best transactions")? - .is_some(); - - let new_transactions: Vec<_> = sim_fb_state - .slice_new_transactions(&sim_info.executed_transactions) - .iter() - .map(|tx| tx.tx_hash()) - .collect::>(); - best_txs.mark_committed(new_transactions); - drop(best_txs); - - if !sim_info.reverted_bundle_tx_hashes.is_empty() { - let hashes = mem::take(&mut sim_info.reverted_bundle_tx_hashes); - self.pool().remove_transactions(hashes); - } - - if exec_cancelled || ctx.cancel().is_cancelled() || block_cancel.is_cancelled() { - break; - } - - if let Err(e) = self.builder_tx().add_builder_txs( + let outcome = self.build_candidate( + CandidateKind::PoolBacked, + budget_targets, + ctx, &state_provider, - &mut sim_info, - &ctx.builder_tx_env(), state, - false, - sim_fb_state.is_first_flashblock(), - sim_fb_state.is_last_flashblock(), - ) { - error!(target: "payload_builder", "Error adding bottom builder txs: {}", e); - } - - if ctx.cancel().is_cancelled() || block_cancel.is_cancelled() { - break; - } - - let total_block_built_start = Instant::now(); - let build_result = ctx - .block_assembly_input() - .map_err(|e| eyre::eyre!("failed to construct block assembly input: {e}")) - .and_then(|input| { - input - .assemble( - state, - Some(&mut sim_fb_state), - &mut sim_info, - &mut sim_state_root_calc, - ctx.metrics.clone(), - ctx.enable_tx_tracking_debug_logs, - ) - .map_err(|e| eyre::eyre!("failed to assemble candidate: {e}")) - }); - let total_block_built_duration = total_block_built_start.elapsed(); - - candidates_evaluated += 1; - - match build_result { - Ok((new_payload, mut fb_payload)) => { - fb_payload.index = flashblock_index; - fb_payload.base = None; - + base_info.clone(), + base_fb_state.clone(), + base_tx_tracker.clone(), + base_state_root_calc.clone(), + block_cancel, + InterruptPolicy::JobAndFlashblock, + )?; + + match outcome { + CandidateOutcome::Built(BuiltCandidate { + new_payload, + fb_payload, + next_fb_state, + carried, + timings, + }) => { + candidates_evaluated += 1; let best_total_fees = best.as_ref().map_or(U256::ZERO, |b| b.total_fees); - let is_new_best = sim_info.total_fees > best_total_fees || best.is_none(); - candidate_span.record("gas_used", sim_info.cumulative_gas_used); - candidate_span.record("total_fees_wei", sim_info.total_fees.to_string()); + let is_new_best = carried.info.total_fees > best_total_fees || best.is_none(); + candidate_span.record("gas_used", carried.info.cumulative_gas_used); + candidate_span.record("total_fees_wei", carried.info.total_fees.to_string()); candidate_span.record("is_best", is_new_best); // Record the first successfully-built candidate's fees as // the improvement baseline (in case the empty-candidate // path failed or didn't run). if first_candidate_fees.is_none() { - first_candidate_fees = Some(sim_info.total_fees); + first_candidate_fees = Some(carried.info.total_fees); } if is_new_best { - let next_flashblock_state = sim_fb_state - .next_after_seal(target_da_for_batch, target_da_footprint_for_batch); let candidate = BestCandidate { - total_fees: sim_info.total_fees, - cache: state.cache.clone(), - transition: state.transition_state.clone(), - info: sim_info, - fb_state: sim_fb_state, - tx_tracker: sim_tx_tracker, + total_fees: carried.info.total_fees, + cache: carried.cache, + transition: carried.transition, + info: carried.info, + fb_state: carried.fb_state, + tx_tracker: carried.tx_tracker, build_duration: candidate_build_start.elapsed(), - transaction_pool_fetch_duration: Some(transaction_pool_fetch_time), - total_block_built_duration: Some(total_block_built_duration), - result: (next_flashblock_state, new_payload, fb_payload), + transaction_pool_fetch_duration: timings.pool_fetch, + total_block_built_duration: Some(timings.assemble), + result: (next_fb_state, new_payload, fb_payload), limiter_snapshot: ctx.address_limiter().snapshot_pending(), candidates_evaluated, candidates_improved: candidates_improved + 1, - state_root_calc: sim_state_root_calc, + state_root_calc: carried.state_root_calc, }; // Publish to shared slot so the main loop can take it // on trigger without awaiting this task. @@ -411,8 +266,16 @@ where shared_best.refresh_metrics(candidates_evaluated, candidates_improved); } } - Err(err) => { + CandidateOutcome::Cancelled(cancelled) => { + cancelled + .carried + .restore(state, info, fb_state, tx_tracker, state_root_calc); + break; + } + CandidateOutcome::SealFailed(failure) => { + candidates_evaluated += 1; ctx.metrics.invalid_built_blocks_count.increment(1); + let err = failure.into_continuous_error(); warn!(target: "payload_builder", ?err, "Candidate seal failed, continuing"); if best.is_some() { shared_best.refresh_metrics(candidates_evaluated, candidates_improved); diff --git a/crates/op-rbuilder/src/builder/mod.rs b/crates/op-rbuilder/src/builder/mod.rs index 535cc04c..d8a9f7b0 100644 --- a/crates/op-rbuilder/src/builder/mod.rs +++ b/crates/op-rbuilder/src/builder/mod.rs @@ -13,6 +13,7 @@ mod assembly; mod best_txs; mod builder_tx; pub(crate) mod cancellation; +mod candidate; mod config; mod context; mod continuous; diff --git a/crates/op-rbuilder/src/builder/payload.rs b/crates/op-rbuilder/src/builder/payload.rs index cd9882be..520d7fe2 100644 --- a/crates/op-rbuilder/src/builder/payload.rs +++ b/crates/op-rbuilder/src/builder/payload.rs @@ -2,9 +2,13 @@ use super::{state_root::StateRootCalculator, wspub::WebSocketPublisher}; use crate::{ builder::{ BuilderConfig, - best_txs::{FlashblockPoolTxCursor, FlashblockTxTracker}, - builder_tx::{BuilderTransactions, reserve_builder_tx_budget}, + best_txs::FlashblockTxTracker, + builder_tx::BuilderTransactions, cancellation::{CancellationReason, FlashblockJobCancellation, PayloadJobCancellation}, + candidate::{ + CandidateKind, CandidateOutcome, CandidateSealFailure, CandidateTimings, + CarriedCandidate, InterruptPolicy, + }, context::{OpPayloadBuilderCtx, OpPayloadJobCtx}, generator::{BuildArguments, PayloadBuilder}, timing::{FlashblockScheduler, compute_slot_offset_ms}, @@ -28,7 +32,6 @@ use reth_optimism_node::{OpBuiltPayload, OpPayloadBuilderAttributes}; use reth_optimism_payload_builder::OpPayloadAttrs; use reth_optimism_primitives::{OpReceipt, OpTransactionSigned}; use reth_payload_builder::PayloadId; -use reth_payload_util::BestPayloadTransactions; use reth_revm::{ State, cached::CachedReads, @@ -37,7 +40,6 @@ use reth_revm::{ }; use reth_tasks::Runtime; use std::{ - mem, ops::Deref, sync::{Arc, atomic::AtomicU64}, time::{Duration, Instant}, @@ -150,6 +152,7 @@ struct FallbackBuildOutput { struct FlashblockBuildOutput { state: BuildState, build_result: eyre::Result>, + timings: CandidateTimings, } /// Result of building one flashblock in the naive loop: either carry the @@ -183,7 +186,7 @@ pub(crate) struct PayloadBuildStats { } impl FlashblocksState { - fn new(target_flashblock_count: u64) -> Self { + pub(crate) fn new(target_flashblock_count: u64) -> Self { Self { target_flashblock_count, ..Default::default() @@ -249,7 +252,7 @@ impl FlashblocksState { ) } - fn with_batch_limits( + pub(crate) fn with_batch_limits( mut self, gas_per_batch: u64, da_per_batch: Option, @@ -904,6 +907,7 @@ where let Some(FlashblockBuildOutput { state: new_state, build_result, + timings, }) = self .run_blocking_flashblock_build(deps.payload_cancel, state, parent_hash, fb_span.clone()) .await? @@ -912,6 +916,27 @@ where }; state = new_state; + if let Some(duration) = timings.pool_fetch { + state + .ctx + .metrics + .transaction_pool_fetch_duration + .record(duration); + state.ctx.metrics.transaction_pool_fetch_gauge.set(duration); + } + if let Some(duration) = timings.execution { + state + .ctx + .metrics + .payload_transaction_simulation_duration + .record(duration); + state + .ctx + .metrics + .payload_transaction_simulation_gauge + .set(duration); + } + // Record span attributes now that we have results fb_span.record("tx_count", state.info.executed_transactions.len() as u64); fb_span.record("gas_used", state.info.cumulative_gas_used); @@ -1156,12 +1181,12 @@ where ) -> eyre::Result { let BuildState { ctx, - mut fb_state, + fb_state, mut info, cache, transition, - mut tx_tracker, - mut state_root_calc, + tx_tracker, + state_root_calc, } = build_state; let state_provider = self.client.state_by_block_hash(parent_hash)?; @@ -1173,176 +1198,71 @@ where evm_state.transition_state = transition; let flashblock_index = fb_state.flashblock_index(); - let mut target_gas_for_batch = fb_state.target_gas_for_batch(); - let mut target_da_for_batch = fb_state.target_da_for_batch(); - let mut target_da_footprint_for_batch = fb_state.target_da_footprint_for_batch(); - info!( target: "payload_builder", block_number = ctx.block_number(), flashblock_index, - target_gas = target_gas_for_batch, + target_gas = fb_state.target_gas_for_batch(), gas_used = info.cumulative_gas_used, - target_da = target_da_for_batch, + target_da = fb_state.target_da_for_batch(), da_used = info.cumulative_da_bytes_used, block_gas_used = ctx.block_gas_limit(), - target_da_footprint = target_da_footprint_for_batch, + target_da_footprint = fb_state.target_da_footprint_for_batch(), "Building flashblock", ); - let flashblock_build_start_time = Instant::now(); - - let flashblock = fb_state.meta(); - let builder_txs = self - .builder_tx - .add_builder_txs( - &state_provider, - &mut info, - &ctx.builder_tx_env(), - &mut evm_state, - true, - flashblock.is_first(), - flashblock.is_last(), - ) - .inspect_err( - |e| error!(target: "payload_builder", error = %e, "Error simulating builder txs"), - ) - .unwrap_or_default(); - - // only reserve builder tx gas / da size that has not been committed yet - // committed builder txs would have counted towards the gas / da used - let max_uncompressed_block_size = reserve_builder_tx_budget( - &builder_txs, - &mut target_gas_for_batch, - &mut target_da_for_batch, - &mut target_da_footprint_for_batch, - info.da_footprint_scalar, - ctx.max_uncompressed_block_size, - info.cumulative_uncompressed_bytes, - ); - - let best_txs_start_time = Instant::now(); - let mut best_txs = FlashblockPoolTxCursor::new(&mut tx_tracker); - best_txs.refresh_iterator( - BestPayloadTransactions::new( - self.pool - .best_transactions_with_attributes(ctx.best_transaction_attributes()), - ), - flashblock_index, - ); - let transaction_pool_fetch_time = best_txs_start_time.elapsed(); - ctx.metrics - .transaction_pool_fetch_duration - .record(transaction_pool_fetch_time); - ctx.metrics - .transaction_pool_fetch_gauge - .set(transaction_pool_fetch_time); - - let tx_execution_start_time = Instant::now(); - ctx.execute_best_transactions( - &mut info, - &mut evm_state, - &mut best_txs, - target_gas_for_batch.min(ctx.block_gas_limit()), - target_da_for_batch, - target_da_footprint_for_batch, - max_uncompressed_block_size, - fb_state.flashblock_index, - ) - .wrap_err("failed to execute best transactions")?; - // Extract last transactions - let new_transactions: Vec<_> = fb_state - .slice_new_transactions(&info.executed_transactions) - .iter() - .map(|tx| tx.tx_hash()) - .collect::>(); - best_txs.mark_committed(new_transactions); - - // Remove reverted bundle txs from the pool so they aren't re-simulated in future blocks - if !info.reverted_bundle_tx_hashes.is_empty() { - let hashes = mem::take(&mut info.reverted_bundle_tx_hashes); - self.pool.remove_transactions(hashes); - } - - // Block cancelled (new FCU, getPayload resolved, or deadline). Skip publishing. - if block_cancel.is_cancelled() { - let cache = std::mem::take(&mut evm_state.cache); - let transition = evm_state.transition_state.take(); - return Ok(FlashblockBuildOutput { - state: BuildState { - ctx, - info, - cache, - transition, - tx_tracker, - fb_state, - state_root_calc, - }, - build_result: Ok(None), - }); - } - - let payload_transaction_simulation_time = tx_execution_start_time.elapsed(); - ctx.metrics - .payload_transaction_simulation_duration - .record(payload_transaction_simulation_time); - ctx.metrics - .payload_transaction_simulation_gauge - .set(payload_transaction_simulation_time); - - let flashblock = fb_state.meta(); - if let Err(e) = self.builder_tx.add_builder_txs( + let flashblock_build_start = Instant::now(); + let budget_targets = self.prepare_candidate_budget( + &ctx, &state_provider, + &mut evm_state, &mut info, - &ctx.builder_tx_env(), + &fb_state, + ); + let outcome = self.build_candidate( + CandidateKind::PoolBacked, + budget_targets, + &ctx, + &state_provider, &mut evm_state, - false, - flashblock.is_first(), - flashblock.is_last(), - ) { - error!(target: "payload_builder", error = %e, "Error simulating builder txs"); - } - - let (new_payload, mut fb_payload) = ctx - .block_assembly_input()? - .assemble( - &mut evm_state, - Some(&mut fb_state), - &mut info, - &mut state_root_calc, - ctx.metrics.clone(), - ctx.enable_tx_tracking_debug_logs, - ) - .inspect_err(|_| ctx.metrics.invalid_built_blocks_count.increment(1)) - .context("failed to build payload")?; - - fb_payload.index = flashblock_index; - fb_payload.base = None; - - // Block canceled (new FCU, getPayload resolved, or deadline). The async outer - // loop owns publishing and re-checks cancellation before every side effect. - if block_cancel.is_cancelled() { - let cache = std::mem::take(&mut evm_state.cache); - let transition = evm_state.transition_state.take(); - return Ok(FlashblockBuildOutput { - state: BuildState { - ctx, - info, - cache, - transition, - tx_tracker, - fb_state, - state_root_calc, - }, - build_result: Ok(None), - }); - } + info, + fb_state, + tx_tracker, + state_root_calc, + block_cancel, + InterruptPolicy::JobOnly, + )?; - // Advance batch budgets for the next flashblock. - let next_flashblock_state = - fb_state.next_after_seal(target_da_for_batch, target_da_footprint_for_batch); + let (carried, build_result, timings) = match outcome { + CandidateOutcome::Built(built) => { + let next_flashblock_state = built.next_fb_state; + let new_payload = built.new_payload; + let fb_payload = built.fb_payload; + let timings = built.timings; + let build_result = + (!block_cancel.is_cancelled()).then_some(BuiltFlashblockOutput { + next_flashblock_state, + new_payload, + fb_payload, + build_duration: flashblock_build_start.elapsed(), + }); + (built.carried, build_result, timings) + } + CandidateOutcome::Cancelled(cancelled) => (cancelled.carried, None, cancelled.timings), + CandidateOutcome::SealFailed(CandidateSealFailure::Input(err)) => return Err(err), + CandidateOutcome::SealFailed(CandidateSealFailure::Assemble(err)) => { + ctx.metrics.invalid_built_blocks_count.increment(1); + return Err(err.wrap_err("failed to build payload")); + } + }; + let CarriedCandidate { + cache, + transition, + info, + fb_state, + tx_tracker, + state_root_calc, + } = carried; - let cache = std::mem::take(&mut evm_state.cache); - let transition = evm_state.transition_state.take(); Ok(FlashblockBuildOutput { state: BuildState { ctx, @@ -1353,12 +1273,8 @@ where fb_state, state_root_calc, }, - build_result: Ok(Some(BuiltFlashblockOutput { - next_flashblock_state, - new_payload, - fb_payload, - build_duration: flashblock_build_start_time.elapsed(), - })), + build_result: Ok(build_result), + timings, }) }