1use crate::tree::{
99 error::{
100 BlockAccessListDecodeError, InsertBlockError, InsertBlockErrorKind, InsertPayloadError,
101 },
102 instrumented_state::{InstrumentedStateProvider, StateProviderMetrics, StateProviderStats},
103 payload_processor::PayloadProcessor,
104 precompile_cache::{CachedPrecompile, CachedPrecompileMetrics, PrecompileCacheMap},
105 txpool_prewarm,
106 types::{InsertPayloadResult, ValidationOutput},
107 CacheWaitDurations, CachedStateProvider, EngineApiMetrics, EngineApiTreeState, ExecutionEnv,
108 PayloadHandle, StateProviderDatabase, TreeConfig, WaitForCaches,
109};
110use alloy_consensus::transaction::{Either, TxHashRef};
111use alloy_eip7928::{
112 bal::{Bal, DecodedBal},
113 BlockAccessList,
114};
115use alloy_eips::{eip1898::BlockWithParent, eip4895::Withdrawal, NumHash};
116use alloy_evm::Evm;
117use alloy_primitives::{
118 map::{AddressMap, B256Set},
119 B256,
120};
121use reth_tasks::LazyHandle;
122
123use crate::tree::{
124 payload_processor::receipt_root_task::{IndexedReceipt, ReceiptRootTaskHandle},
125 state_root_strategy::{
126 DefaultStateRootStrategy, LazyHashedPostState, PayloadStateRootHandle,
127 PayloadStateRootJobContext, StateRootHintStream, StateRootJobContext, StateRootStrategy,
128 StateRootUpdateStream,
129 },
130};
131use alloy_consensus::constants::KECCAK_EMPTY;
132use alloy_primitives::Address;
133use reth_chain_state::{CanonicalInMemoryState, ExecutedBlock, ExecutionTimingStats};
134use reth_consensus::{ConsensusError, FullConsensus, ReceiptRootBloom};
135use reth_engine_primitives::{
136 ConfigureEngineEvm, ExecutableTxIterator, ExecutionPayload, InvalidBlockHook, PayloadValidator,
137};
138use reth_errors::{BlockExecutionError, BlockValidationError, ProviderResult};
139use reth_evm::{
140 block::BlockExecutor, execute::ExecutableTxFor, ConfigureEvm, EvmEnvFor, ExecutionCtxFor,
141 OnStateHook, SpecFor,
142};
143use reth_execution_cache::{CacheFillMode, CacheStats};
144use reth_network_p2p::full_block::SealedBlockWithAccessList;
145use reth_payload_builder::{PayloadBuilderLease, PayloadBuilderResources};
146use reth_payload_primitives::{
147 BuiltPayload, BuiltPayloadExecutedBlock, InvalidPayloadAttributesError, NewPayloadError,
148 PayloadTypes,
149};
150use reth_primitives_traits::{
151 AlloyBlockHeader, BlockBody, BlockTy, FastInstant as Instant, GotExpected, NodePrimitives,
152 RecoveredBlock, SealedBlock, SealedHeader, SignerRecoverable,
153};
154use reth_provider::{
155 BlockExecutionOutput, BlockHashReader, BlockReader, ChangeSetReader, DatabaseProviderFactory,
156 DatabaseProviderROFactory, HashedPostStateProvider, HistoryReader, ProviderError,
157 PruneCheckpointReader, StageCheckpointReader, StateProvider, StateProviderBox,
158 StateProviderFactory, StateReader, StateRootProvider, StorageChangeSetReader,
159 StorageSettingsCache,
160};
161use reth_revm::db::{states::bundle_state::BundleRetention, BundleAccount, State};
162use reth_storage_overlay::{OverlayManager, OverlayStateProviderFactory};
163use reth_trie::{
164 hashed_cursor::HashedCursorFactory, trie_cursor::TrieCursorFactory, updates::TrieUpdates,
165 HashedPostState, KeccakKeyHasher, LazyTrieData,
166};
167use std::{
168 sync::{
169 atomic::{AtomicUsize, Ordering},
170 Arc,
171 },
172 time::Duration,
173};
174use tracing::{debug, debug_span, error, info, instrument, trace, warn, Level, Span};
175
176pub use crate::tree::types::ValidationOutcome;
177
178const MAX_EXPECTED_GAS_LIMIT_MULTIPLIER: u64 = 2;
181
182const DEFERRED_TRIE_WORKER_NAME: &str = "deferred-trie";
184
185type ReceiptRootSender<N> =
186 crossbeam_channel::Sender<IndexedReceipt<<N as NodePrimitives>::Receipt>>;
187type ReceiptRootReceiver = tokio::sync::oneshot::Receiver<(B256, alloy_primitives::Bloom)>;
188
189pub struct TreeCtx<'a, N: NodePrimitives> {
194 state: &'a mut EngineApiTreeState<N>,
196 canonical_in_memory_state: &'a CanonicalInMemoryState<N>,
198}
199
200impl<'a, N: NodePrimitives> std::fmt::Debug for TreeCtx<'a, N> {
201 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
202 f.debug_struct("TreeCtx")
203 .field("state", &"EngineApiTreeState")
204 .field("canonical_in_memory_state", &self.canonical_in_memory_state)
205 .finish()
206 }
207}
208
209impl<'a, N: NodePrimitives> TreeCtx<'a, N> {
210 pub const fn new(
212 state: &'a mut EngineApiTreeState<N>,
213 canonical_in_memory_state: &'a CanonicalInMemoryState<N>,
214 ) -> Self {
215 Self { state, canonical_in_memory_state }
216 }
217}
218
219impl<'a, N: NodePrimitives> TreeCtx<'a, N> {
220 pub const fn state(&self) -> &EngineApiTreeState<N> {
222 &*self.state
223 }
224
225 pub const fn state_mut(&mut self) -> &mut EngineApiTreeState<N> {
227 self.state
228 }
229
230 pub const fn canonical_in_memory_state(&self) -> &'a CanonicalInMemoryState<N> {
232 self.canonical_in_memory_state
233 }
234}
235
236struct JitPauseGuard<Evm: ConfigureEvm>(Evm);
242
243impl<Evm: ConfigureEvm> JitPauseGuard<Evm> {
244 fn new(evm_config: &Evm) -> Self {
245 if let Some(jit_backend) = evm_config.jit_backend() {
246 jit_backend.pause();
247 }
248 Self(evm_config.clone())
249 }
250}
251
252impl<Evm: ConfigureEvm> Drop for JitPauseGuard<Evm> {
253 fn drop(&mut self) {
254 if let Some(jit_backend) = self.0.jit_backend() {
255 jit_backend.resume();
256 }
257 }
258}
259
260#[derive(derive_more::Debug)]
268pub struct BasicEngineValidator<P, Evm, V>
269where
270 Evm: ConfigureEvm,
271{
272 provider: P,
274 consensus: Arc<dyn FullConsensus<Evm::Primitives>>,
276 evm_config: Evm,
278 config: TreeConfig,
280 payload_processor: PayloadProcessor<Evm>,
282 precompile_cache_map: PrecompileCacheMap<SpecFor<Evm>>,
284 precompile_cache_metrics: AddressMap<CachedPrecompileMetrics>,
286 #[debug(skip)]
288 invalid_block_hook: Box<dyn InvalidBlockHook<Evm::Primitives>>,
289 metrics: EngineApiMetrics,
291 validator: V,
293 runtime: reth_tasks::Runtime,
295 overlay_manager: OverlayManager<Evm::Primitives>,
297 #[debug(skip)]
299 state_root_strategy: Arc<dyn StateRootStrategy<Evm::Primitives, P, Evm>>,
300 #[debug(skip)]
304 txpool_prewarm: Option<txpool_prewarm::Handle<Evm::Primitives, P, Evm>>,
305 bal_hash_buf: Vec<u8>,
307}
308
309impl<N, P, Evm, V> BasicEngineValidator<P, Evm, V>
310where
311 N: NodePrimitives,
312 P: DatabaseProviderFactory<
313 Provider: BlockReader
314 + BlockHashReader
315 + StageCheckpointReader
316 + PruneCheckpointReader
317 + ChangeSetReader
318 + StorageChangeSetReader
319 + StorageSettingsCache
320 + HistoryReader
321 + 'static,
322 > + BlockReader<Header = N::BlockHeader>
323 + ChangeSetReader
324 + StateProviderFactory
325 + StateReader
326 + Clone
327 + 'static,
328 OverlayStateProviderFactory<P, N>: DatabaseProviderROFactory<
329 Provider: TrieCursorFactory
330 + HashedCursorFactory
331 + HashedPostStateProvider
332 + StateRootProvider
333 + StateProvider
334 + Send,
335 > + Clone
336 + 'static,
337 Evm: ConfigureEvm<Primitives = N> + 'static,
338{
339 #[expect(clippy::too_many_arguments)]
341 pub fn new(
342 provider: P,
343 consensus: Arc<dyn FullConsensus<N>>,
344 evm_config: Evm,
345 validator: V,
346 config: TreeConfig,
347 invalid_block_hook: Box<dyn InvalidBlockHook<N>>,
348 overlay_manager: OverlayManager<N>,
349 runtime: reth_tasks::Runtime,
350 ) -> Self {
351 let precompile_cache_map = PrecompileCacheMap::default();
352 let payload_processor = PayloadProcessor::new(
353 runtime.clone(),
354 evm_config.clone(),
355 &config,
356 precompile_cache_map.clone(),
357 );
358 Self {
359 provider,
360 consensus,
361 evm_config,
362 payload_processor,
363 precompile_cache_map,
364 precompile_cache_metrics: AddressMap::default(),
365 config,
366 invalid_block_hook,
367 metrics: EngineApiMetrics::default(),
368 validator,
369 runtime,
370 overlay_manager,
371 state_root_strategy: Arc::new(DefaultStateRootStrategy::default()),
372 txpool_prewarm: None,
373 bal_hash_buf: Vec::new(),
374 }
375 }
376
377 pub fn with_state_root_strategy(
379 mut self,
380 state_root_strategy: Arc<dyn StateRootStrategy<N, P, Evm>>,
381 ) -> Self {
382 self.state_root_strategy = state_root_strategy;
383 self
384 }
385
386 pub fn with_txpool_prewarming(
388 mut self,
389 source: impl crate::tree::TxPoolPrewarmSource<N> + 'static,
390 ) -> Self {
391 self.txpool_prewarm = Some(txpool_prewarm::Handle::spawn(
392 &self.runtime,
393 Arc::new(source),
394 self.evm_config.clone(),
395 ));
396 self
397 }
398
399 #[instrument(level = "debug", target = "engine::tree::payload_validator", skip_all)]
401 pub fn convert_to_block<T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>>(
402 &self,
403 input: BlockOrPayload<T>,
404 ) -> Result<SealedBlock<N::Block>, NewPayloadError>
405 where
406 V: PayloadValidator<T, Block = N::Block>,
407 {
408 match input {
409 BlockOrPayload::Payload(payload) => self.validator.convert_payload_to_block(payload),
410 BlockOrPayload::Block(block) => Ok(block.split().0),
411 }
412 }
413
414 pub fn evm_env_for<T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>>(
416 &self,
417 input: &BlockOrPayload<T>,
418 ) -> Result<EvmEnvFor<Evm>, Evm::Error>
419 where
420 V: PayloadValidator<T, Block = N::Block>,
421 Evm: ConfigureEngineEvm<T::ExecutionData, Primitives = N>,
422 {
423 match input {
424 BlockOrPayload::Payload(payload) => Ok(self.evm_config.evm_env_for_payload(payload)?),
425 BlockOrPayload::Block(block) => Ok(self.evm_config.evm_env(block.header())?),
426 }
427 }
428
429 pub fn tx_iterator_for<'a, T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>>(
431 &'a self,
432 input: &'a BlockOrPayload<T>,
433 ) -> Result<impl ExecutableTxIterator<Evm>, NewPayloadError>
434 where
435 V: PayloadValidator<T, Block = N::Block>,
436 Evm: ConfigureEngineEvm<T::ExecutionData, Primitives = N>,
437 {
438 Ok(match input {
439 BlockOrPayload::Payload(payload) => {
440 let iter = self
441 .evm_config
442 .tx_iterator_for_payload(payload)
443 .map_err(NewPayloadError::other)?;
444 Either::Left(iter)
445 }
446 BlockOrPayload::Block(block) => {
447 let txs = block.body().clone_transactions();
448 let convert = |tx: N::SignedTx| tx.try_into_recovered();
449 Either::Right((txs, convert))
450 }
451 })
452 }
453
454 pub fn execution_ctx_for<'a, T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>>(
456 &self,
457 input: &'a BlockOrPayload<T>,
458 ) -> Result<ExecutionCtxFor<'a, Evm>, Evm::Error>
459 where
460 V: PayloadValidator<T, Block = N::Block>,
461 Evm: ConfigureEngineEvm<T::ExecutionData, Primitives = N>,
462 {
463 match input {
464 BlockOrPayload::Payload(payload) => Ok(self.evm_config.context_for_payload(payload)?),
465 BlockOrPayload::Block(block) => Ok(self.evm_config.context_for_block(block)?),
466 }
467 }
468
469 #[instrument(
477 level = "debug",
478 target = "engine::tree::payload_validator",
479 skip_all,
480 fields(
481 parent = ?input.parent_hash(),
482 type_name = ?input.type_name(),
483 )
484 )]
485 pub fn validate_block_with_state<T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>>(
486 &mut self,
487 input: BlockOrPayload<T>,
488 mut ctx: TreeCtx<'_, N>,
489 ) -> InsertPayloadResult<N>
490 where
491 V: PayloadValidator<T, Block = N::Block> + Clone,
492 Evm: ConfigureEngineEvm<T::ExecutionData, Primitives = N>,
493 {
494 let parent_hash = input.parent_hash();
495 let _txpool_pause = self.txpool_prewarm.as_ref().map(txpool_prewarm::Handle::pause);
496 let txpool_snapshot =
497 self.txpool_prewarm.as_ref().and_then(|prewarmer| prewarmer.snapshot(parent_hash));
498 let _jit_pause = JitPauseGuard::new(&self.evm_config);
499
500 let parent_block = match self.sealed_header_by_hash(parent_hash, ctx.state()) {
503 Ok(Some(parent_block)) => parent_block,
504 Ok(None) => {
505 return Err(InsertBlockError::new(
506 self.convert_to_block(input)?,
507 ProviderError::HeaderNotFound(parent_hash.into()).into(),
508 )
509 .into())
510 }
511 Err(e) => {
512 return Err(InsertBlockError::new(self.convert_to_block(input)?, e.into()).into())
513 }
514 };
515
516 let validated_block = self.spawn_convert_and_validate(&input, parent_block.clone());
520
521 macro_rules! ensure_ok {
524 ($expr:expr) => {
525 match $expr {
526 Ok(val) => val,
527 Err(e) => {
528 let block = validated_block.try_into_inner().expect("sole handle")?;
529 return Err(InsertBlockError::new(block, e.into()).into())
530 }
531 }
532 };
533 }
534
535 macro_rules! ensure_ok_post_block {
537 ($expr:expr, $block:expr) => {
538 match $expr {
539 Ok(val) => val,
540 Err(e) => {
541 return Err(
542 InsertBlockError::new($block.into_sealed_block(), e.into()).into()
543 )
544 }
545 }
546 };
547 }
548
549 if input.gas_limit() >
552 parent_block.gas_limit().saturating_mul(MAX_EXPECTED_GAS_LIMIT_MULTIPLIER)
553 {
554 if validated_block.get().is_err() {
556 return Err(validated_block
557 .try_into_inner()
558 .expect("sole handle")
559 .expect_err("Err result checked"))
560 }
561 }
562
563 trace!(target: "engine::tree::payload_validator", "Fetching block state provider");
564 let _enter =
565 debug_span!(target: "engine::tree::payload_validator", "state_provider").entered();
566 let Some(state_provider_factory) =
567 ensure_ok!(self.overlay_state_provider_factory(parent_hash, ctx.state()))
568 else {
569 return Err(InsertBlockError::new(
571 validated_block.try_into_inner().expect("sole handle")?,
572 ProviderError::HeaderNotFound(parent_hash.into()).into(),
573 )
574 .into())
575 };
576 drop(_enter);
577
578 let evm_env = debug_span!(target: "engine::tree::payload_validator", "evm_env")
579 .in_scope(|| self.evm_env_for(&input))
580 .map_err(NewPayloadError::other)?;
581
582 let decoded_bal =
585 ensure_ok!(input.try_decoded_access_list().map_err(BlockAccessListDecodeError::new))
586 .map(Arc::new);
587
588 if let Some(decoded_bal) = decoded_bal.as_deref() {
589 ensure_ok!(decoded_bal
591 .as_bal()
592 .validate_gas_limit(input.gas_limit())
593 .map_err(ConsensusError::from));
594 }
595
596 let env = ExecutionEnv {
597 evm_env,
598 hash: input.hash(),
599 parent_hash: input.parent_hash(),
600 parent_state_root: parent_block.state_root(),
601 transaction_count: input.transaction_count(),
602 gas_used: input.gas_used(),
603 withdrawals: input.withdrawals().map(|w| w.to_vec()),
604 decoded_bal: decoded_bal.as_ref().map(Arc::clone),
605 txpool_snapshot: txpool_snapshot.clone(),
606 };
607
608 let txs = self.tx_iterator_for(&input)?;
610
611 let parallel_bal_execution = ensure_ok!(self.bal_path_eligible(env.decoded_bal.as_deref()));
612
613 let mut state_root_job =
615 ensure_ok!(self.state_root_strategy.prepare(StateRootJobContext::new(
616 &self.runtime,
617 &self.overlay_manager,
618 &env,
619 &parent_block,
620 state_provider_factory.clone(),
621 &self.config,
622 parallel_bal_execution,
623 ctx.state_mut(),
624 )));
625 let state_root_job_name = state_root_job.name();
626
627 debug!(
628 target: "engine::tree::payload_validator",
629 strategy = state_root_job_name,
630 "Prepared state root job"
631 );
632
633 let execution_state_hook = state_root_job.take_execution_hook();
636 let hint_stream = state_root_job.take_hint_stream();
639 let hashed_update_stream = state_root_job.take_hashed_update_stream();
640
641 let mut handle = ensure_ok!(self.spawn_payload_processor(
643 env.clone(),
644 txs,
645 state_provider_factory.clone(),
646 hint_stream,
647 hashed_update_stream,
648 parallel_bal_execution,
649 ));
650
651 let slow_block_enabled = self.config.slow_block_threshold().is_some();
653 let cache_stats = slow_block_enabled.then(|| Arc::new(CacheStats::default()));
654 let instrument_state_provider = slow_block_enabled || self.config.state_provider_metrics();
655 let state_provider_metrics =
656 instrument_state_provider.then(|| StateProviderMetrics::with_source("engine"));
657 let state_provider_stats =
658 instrument_state_provider.then(|| Arc::new(StateProviderStats::default()));
659 let execution_cache = handle.caches().map(|caches| (caches, handle.cache_metrics()));
660
661 let make_state_provider = |fill_on_miss: bool| -> ProviderResult<StateProviderBox> {
681 let provider = state_provider_factory.database_provider_ro()?;
682 let mut provider = if let Some((caches, cache_metrics)) = &execution_cache {
683 let fill_mode = if fill_on_miss {
684 CacheFillMode::FillOnMiss
685 } else {
686 CacheFillMode::LookupOnly
687 };
688 Box::new(
689 CachedStateProvider::new_with_mode(
690 provider,
691 caches.clone(),
692 fill_mode,
693 cache_metrics.clone(),
694 cache_stats.clone(),
695 )
696 .with_txpool_snapshot(txpool_snapshot.clone()),
697 ) as StateProviderBox
698 } else {
699 Box::new(provider) as StateProviderBox
700 };
701
702 if instrument_state_provider {
703 let stats = state_provider_stats
704 .as_ref()
705 .expect("instrumented state provider requires shared stats");
706 let metrics = state_provider_metrics
707 .as_ref()
708 .expect("instrumented state provider requires metrics");
709 provider = Box::new(InstrumentedStateProvider::with_stats(
710 provider,
711 metrics.clone(),
712 Arc::clone(stats),
713 ));
714 }
715
716 Ok(provider)
717 };
718
719 let execute_block_start = Instant::now();
723 let execution_result = if parallel_bal_execution {
724 self.execute_block_bal(env, &input, &handle, &make_state_provider)
725 } else {
726 let state_provider = make_state_provider(false);
727 match state_provider {
728 Ok(state_provider) => self.execute_block(
729 state_provider,
730 env,
731 &input,
732 &mut handle,
733 execution_state_hook,
734 ),
735 Err(err) => Err(err.into()),
736 }
737 };
738 let execution_duration = execute_block_start.elapsed();
739 if let (Some(metrics), Some(stats)) = (&state_provider_metrics, &state_provider_stats) {
740 metrics.record_totals(stats);
741 }
742 let (output, senders, receipt_root_rx, built_bal) = ensure_ok!(execution_result);
743
744 handle.stop_prewarming_execution();
746
747 let output = Arc::new(output);
751
752 let valid_block_tx = handle.terminate_caching(Some(output.clone()));
755
756 let hashed_state_output = output.clone();
760 let mut hashed_state_rx = state_root_job.take_hashed_state_rx();
761 let mut hashed_state: LazyHashedPostState =
762 self.runtime.spawn_blocking_named("hash-post-state", move || {
763 let _span = debug_span!(
764 target: "engine::tree::payload_validator",
765 "hashed_post_state",
766 )
767 .entered();
768 if let Some(Ok(state)) = hashed_state_rx.as_mut().map(|rx| rx.recv()) {
769 state
770 } else {
771 Arc::new(HashedPostState::from_bundle_state::<KeccakKeyHasher>(
772 hashed_state_output.state.state(),
773 ))
774 }
775 });
776
777 let block = validated_block.try_into_inner().expect("sole handle")?;
778 let block = block.with_senders(senders);
779
780 let receipt_root_bloom = {
782 let _enter = debug_span!(
783 target: "engine::tree::payload_validator",
784 "wait_receipt_root",
785 )
786 .entered();
787
788 receipt_root_rx
789 .blocking_recv()
790 .inspect_err(|_| {
791 tracing::error!(
792 target: "engine::tree::payload_validator",
793 "Receipt root task dropped sender without result, receipt root calculation likely aborted"
794 );
795 })
796 .ok()
797 };
798
799 ensure_ok_post_block!(
800 self.validate_post_execution(
801 &block,
802 &parent_block,
803 &output,
804 &mut ctx,
805 receipt_root_bloom,
806 built_bal
807 ),
808 block
809 );
810
811 let mut hashed_state_validate_result = debug_span!(
812 target: "engine::tree::payload_validator",
813 "validate_block_post_execution_with_hashed_state"
814 )
815 .in_scope(|| {
816 self.validator.validate_block_post_execution_with_hashed_state(
817 || hashed_state.get(),
818 &block,
819 &parent_block,
820 || {
821 state_provider_factory
822 .database_provider_ro()
823 .map(|provider| Box::new(provider) as _)
824 },
825 )
826 });
827
828 let root_start = Instant::now();
829 let root_outcome = ensure_ok_post_block!(
830 state_root_job.finish(&block, output.clone(), &hashed_state),
831 block
832 );
833 let root_elapsed = root_start.elapsed();
834
835 info!(
836 target: "engine::tree::payload_validator",
837 strategy = state_root_job_name,
838 state_root = ?root_outcome.state_root,
839 elapsed = ?root_elapsed,
840 "State root job finished"
841 );
842
843 let state_root = root_outcome.state_root;
844 let trie_output = root_outcome.trie_updates;
845
846 if let Some(refreshed) = root_outcome.hashed_state {
850 hashed_state = LazyHandle::ready(refreshed);
851 hashed_state_validate_result = debug_span!(
852 target: "engine::tree::payload_validator",
853 "validate_block_post_execution_with_hashed_state"
854 )
855 .in_scope(|| {
856 self.validator.validate_block_post_execution_with_hashed_state(
857 || hashed_state.get(),
858 &block,
859 &parent_block,
860 || {
861 state_provider_factory
862 .database_provider_ro()
863 .map(|provider| Box::new(provider) as _)
864 },
865 )
866 });
867 }
868
869 if let Err(err) = hashed_state_validate_result {
870 if err.is_validation_error() {
871 self.on_invalid_block(&parent_block, &block, &output, None, ctx.state_mut());
872 }
873 return Err(InsertBlockError::new(block.into_sealed_block(), err).into())
874 }
875
876 self.metrics.block_validation.record_state_root(&trie_output, root_elapsed.as_secs_f64());
877 self.metrics
878 .record_state_root_gas_bucket(block.header().gas_used(), root_elapsed.as_secs_f64());
879 debug!(target: "engine::tree::payload_validator", ?root_elapsed, "Calculated state root");
880
881 if state_root != block.header().state_root() {
883 self.on_invalid_block(
885 &parent_block,
886 &block,
887 &output,
888 Some((&trie_output, state_root)),
889 ctx.state_mut(),
890 );
891 let block_state_root = block.header().state_root();
892 return Err(InsertBlockError::new(
893 block.into_sealed_block(),
894 ConsensusError::BodyStateRootDiff(
895 GotExpected { got: state_root, expected: block_state_root }.into(),
896 )
897 .into(),
898 )
899 .into())
900 }
901
902 let timing_stats = state_provider_stats.filter(|_| slow_block_enabled).map(|stats| {
903 self.calculate_timing_stats(
904 &block,
905 stats,
906 cache_stats,
907 &output,
908 execution_duration,
909 root_elapsed,
910 )
911 });
912
913 if let Some(valid_block_tx) = valid_block_tx {
914 let _ = valid_block_tx.send(());
915 }
916
917 let executed_block =
918 self.spawn_deferred_trie_task(Arc::new(block), output, hashed_state, trie_output);
919 let raw_bal = decoded_bal.map(|decoded_bal| decoded_bal.as_raw_bal().clone());
920 Ok(ValidationOutput::new(executed_block, timing_stats).with_raw_bal(raw_bal))
921 }
922
923 #[expect(clippy::type_complexity)]
926 pub fn spawn_convert_and_validate<T>(
927 &self,
928 input: &BlockOrPayload<T>,
929 parent: SealedHeader<N::BlockHeader>,
930 ) -> LazyHandle<Result<SealedBlock<N::Block>, InsertPayloadError<N::Block>>>
931 where
932 T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>,
933 V: PayloadValidator<T, Block = N::Block> + Clone,
934 {
935 let input = input.clone();
936 let validator = self.validator.clone();
937 let consensus = self.consensus.clone();
938 let parent_span = Span::current();
939 self.runtime.spawn_blocking_named("payload-convert", move || {
940 let _span = debug_span!(
941 target: "engine::tree::payload_validator",
942 parent: parent_span,
943 "convert_and_validate",
944 )
945 .entered();
946 let block = match input {
947 BlockOrPayload::Block(block) => block.split().0,
948 BlockOrPayload::Payload(payload) => {
949 validator.convert_payload_to_block(payload)?
950 }
951 };
952
953 if let Err(e) = consensus.validate_header(block.sealed_header()) {
954 error!(target: "engine::tree::payload_validator", ?block, "Failed to validate header {}: {e}", block.hash());
955 return Err(InsertBlockError::consensus_error(e, block).into())
956 }
957
958 let _enter = debug_span!(target: "engine::tree::payload_validator", "validate_header_against_parent").entered();
960 if let Err(e) = consensus.validate_header_against_parent(block.sealed_header(), &parent)
961 {
962 warn!(target: "engine::tree::payload_validator", ?block, "Failed to validate header {} against parent: {e}", block.hash());
963 return Err(InsertBlockError::consensus_error(e, block).into())
964 }
965 drop(_enter);
966
967 if let Err(e) =
968 consensus.validate_block_pre_execution_with_tx_root(&block, None)
969 {
970 error!(target: "engine::tree::payload_validator", ?block, "Failed to validate block {}: {e}", block.hash());
971 return Err(InsertBlockError::consensus_error(e, block).into())
972 }
973
974 Ok(block)
975 })
976 }
977
978 fn sealed_header_by_hash(
980 &self,
981 hash: B256,
982 state: &EngineApiTreeState<N>,
983 ) -> ProviderResult<Option<SealedHeader<N::BlockHeader>>> {
984 let header = state.tree_state.sealed_header_by_hash(&hash);
986
987 if header.is_some() {
988 Ok(header)
989 } else {
990 self.provider.sealed_header_by_hash(hash)
991 }
992 }
993
994 #[instrument(level = "debug", target = "engine::tree::payload_validator", skip_all)]
1002 #[expect(clippy::type_complexity)]
1003 fn execute_block<S, Err, T>(
1004 &mut self,
1005 state_provider: S,
1006 env: ExecutionEnv<Evm>,
1007 input: &BlockOrPayload<T>,
1008 handle: &mut PayloadHandle<impl ExecutableTxFor<Evm>, Err, N::Receipt>,
1009 state_hook: Option<Box<dyn OnStateHook + 'static>>,
1010 ) -> Result<
1011 (
1012 BlockExecutionOutput<N::Receipt>,
1013 Vec<Address>,
1014 ReceiptRootReceiver,
1015 Option<BlockAccessList>,
1016 ),
1017 InsertBlockErrorKind,
1018 >
1019 where
1020 S: StateProvider + Send,
1021 Err: core::error::Error + Send + Sync + 'static,
1022 V: PayloadValidator<T, Block = N::Block>,
1023 T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>,
1024 Evm: ConfigureEngineEvm<T::ExecutionData, Primitives = N>,
1025 {
1026 debug!(target: "engine::tree::payload_validator", "Executing block");
1027
1028 let has_bal = input.has_block_access_list();
1029 let mut db = debug_span!(target: "engine::tree", "build_state_db").in_scope(|| {
1030 State::builder()
1031 .with_database(StateProviderDatabase::new(state_provider))
1032 .with_bundle_update()
1033 .with_bal_builder_if(has_bal)
1034 .build()
1035 });
1036
1037 let (spec_id, mut executor) = {
1038 let _span = debug_span!(target: "engine::tree", "create_evm").entered();
1039 let spec_id = *env.evm_env.spec_id();
1040 let evm_config = self.evm_config.clone().with_jit_support();
1041 let evm = evm_config.evm_with_env(&mut db, env.evm_env);
1042 let ctx = self
1043 .execution_ctx_for(input)
1044 .map_err(|e| InsertBlockErrorKind::Other(Box::new(e)))?;
1045 let executor = self.evm_config.create_executor(evm, ctx);
1046 (spec_id, executor)
1047 };
1048
1049 if !self.config.precompile_cache_disabled() {
1050 let _span = debug_span!(target: "engine::tree", "setup_precompile_cache").entered();
1051 executor.evm_mut().precompiles_mut().map_cacheable_precompiles(
1052 |address, precompile| {
1053 let metrics = self
1054 .precompile_cache_metrics
1055 .entry(*address)
1056 .or_insert_with(|| CachedPrecompileMetrics::new_with_address(*address))
1057 .clone();
1058 CachedPrecompile::wrap(
1059 precompile,
1060 self.precompile_cache_map.cache_for_address(*address),
1061 spec_id,
1062 Some(metrics),
1063 )
1064 },
1065 );
1066 }
1067
1068 let transaction_count = input.transaction_count();
1069 let (receipt_tx, result_rx) = self.spawn_receipt_root_task(transaction_count);
1070 let executed_tx_index = Arc::clone(handle.executed_tx_index());
1071 executor.evm_mut().db_mut().set_state_hook(state_hook);
1072
1073 let execution_start = Instant::now();
1074
1075 let (executor, senders) = self.execute_transactions(
1077 executor,
1078 transaction_count,
1079 handle.iter_transactions(),
1080 &receipt_tx,
1081 &executed_tx_index,
1082 has_bal,
1083 )?;
1084 drop(receipt_tx);
1085
1086 let post_exec_start = Instant::now();
1088 let (_evm, result) = debug_span!(target: "engine::tree", "BlockExecutor::finish")
1089 .in_scope(|| executor.finish())
1090 .map(|(evm, result)| (evm.into_db(), result))?;
1091 self.metrics.record_post_execution(post_exec_start.elapsed());
1092
1093 debug_span!(target: "engine::tree", "merge_transitions")
1095 .in_scope(|| db.merge_transitions(BundleRetention::Reverts));
1096
1097 let built_bal = if has_bal { db.take_built_alloy_bal() } else { None };
1098 let output = BlockExecutionOutput { result, state: db.take_bundle() };
1099
1100 let execution_duration = execution_start.elapsed();
1101 self.metrics.record_block_execution(&output, execution_duration);
1102 self.metrics.record_block_execution_gas_bucket(output.result.gas_used, execution_duration);
1103 debug!(target: "engine::tree::payload_validator", elapsed = ?execution_duration, "Executed block");
1104
1105 Ok((output, senders, result_rx, built_bal))
1106 }
1107
1108 fn bal_path_eligible(&self, bal: Option<&DecodedBal>) -> Result<bool, InsertBlockErrorKind> {
1117 let has_bal = bal.is_some();
1118 let parallel_execution = has_bal && !self.config.disable_bal_parallel_execution();
1119 if parallel_execution && self.config.disable_bal_parallel_state_root() {
1120 return Err(InsertBlockErrorKind::Other(
1121 "disabling parallel state root is impossible when parallel execution is enabled"
1122 .into(),
1123 ));
1124 }
1125
1126 Ok(parallel_execution)
1127 }
1128
1129 #[instrument(level = "debug", target = "engine::tree::payload_validator", skip_all)]
1139 #[expect(clippy::type_complexity)]
1140 fn execute_block_bal<Tx, Err, MakeStateProvider, T>(
1141 &self,
1142 env: ExecutionEnv<Evm>,
1143 input: &BlockOrPayload<T>,
1144 handle: &PayloadHandle<Tx, Err, N::Receipt>,
1145 make_state_provider: &MakeStateProvider,
1146 ) -> Result<
1147 (
1148 BlockExecutionOutput<N::Receipt>,
1149 Vec<Address>,
1150 ReceiptRootReceiver,
1151 Option<BlockAccessList>,
1152 ),
1153 InsertBlockErrorKind,
1154 >
1155 where
1156 Tx: ExecutableTxFor<Evm> + Send,
1157 Err: core::error::Error + Send + Sync + 'static,
1158 MakeStateProvider: Fn(bool) -> ProviderResult<StateProviderBox> + Sync,
1159 Evm: ConfigureEngineEvm<T::ExecutionData, Primitives = N>,
1160 T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>,
1161 V: PayloadValidator<T, Block = N::Block>,
1162 {
1163 debug!(target: "engine::tree::payload_validator", "Executing block via BAL path");
1164
1165 let (receipt_tx, result_rx) = self.spawn_receipt_root_task(env.transaction_count);
1166 let input_bal = env.decoded_bal.ok_or_else(|| {
1167 InsertBlockErrorKind::Other("BAL execute path: no decoded BAL available".into())
1168 })?;
1169
1170 let make_db = |fill_on_miss| {
1171 let provider = make_state_provider(fill_on_miss)
1172 .map_err(crate::tree::payload_processor::bal::BalExecutionError::Provider)?;
1173 Ok(StateProviderDatabase::new(provider))
1174 };
1175 let execution_start = Instant::now();
1176 let ctx =
1177 self.execution_ctx_for(input).map_err(|e| InsertBlockErrorKind::Other(Box::new(e)))?;
1178 let (output, senders, built_bal) = crate::tree::payload_processor::bal::execute_block(
1179 &self.runtime,
1180 &self.evm_config,
1181 &make_db,
1182 input_bal,
1183 env.evm_env,
1184 ctx,
1185 env.transaction_count,
1186 handle.clone_transaction_receiver(),
1187 receipt_tx,
1188 )?;
1189 let execution_duration = execution_start.elapsed();
1190
1191 self.metrics.record_block_execution(&output, execution_duration);
1192 self.metrics.record_block_execution_gas_bucket(output.result.gas_used, execution_duration);
1193 debug!(
1194 target: "engine::tree::payload_validator",
1195 elapsed = ?execution_duration,
1196 "Executed block via BAL path",
1197 );
1198
1199 Ok((output, senders, result_rx, Some(built_bal)))
1200 }
1201
1202 fn spawn_receipt_root_task(
1203 &self,
1204 receipts_len: usize,
1205 ) -> (ReceiptRootSender<N>, ReceiptRootReceiver) {
1206 let (receipt_tx, receipt_rx) = crossbeam_channel::unbounded();
1208 let (result_tx, result_rx) = tokio::sync::oneshot::channel();
1209 let task_handle = ReceiptRootTaskHandle::new(receipt_rx, result_tx);
1210 self.runtime.spawn_blocking_named("receipt-root", move || task_handle.run(receipts_len));
1211
1212 (receipt_tx, result_rx)
1213 }
1214
1215 fn execute_transactions<'a, E, Tx, InnerTx, Err, DB>(
1225 &self,
1226 mut executor: E,
1227 transaction_count: usize,
1228 transactions: impl Iterator<Item = Result<Tx, Err>>,
1229 receipt_tx: &crossbeam_channel::Sender<IndexedReceipt<N::Receipt>>,
1230 executed_tx_index: &AtomicUsize,
1231 has_bal: bool,
1232 ) -> Result<(E, Vec<Address>), BlockExecutionError>
1233 where
1234 E: BlockExecutor<Receipt = N::Receipt, Evm: alloy_evm::Evm<DB = &'a mut State<DB>>>,
1235 Tx: alloy_evm::block::ExecutableTx<E> + alloy_evm::RecoveredTx<InnerTx>,
1236 InnerTx: TxHashRef,
1237 DB: revm::Database + 'a,
1238 Err: core::error::Error + Send + Sync + 'static,
1239 {
1240 let mut senders = Vec::with_capacity(transaction_count);
1241
1242 let pre_exec_start = Instant::now();
1244 debug_span!(target: "engine::tree", "pre_execution")
1245 .in_scope(|| executor.apply_pre_execution_changes())?;
1246 self.metrics.record_pre_execution(pre_exec_start.elapsed());
1247
1248 if has_bal {
1250 executor.evm_mut().db_mut().bump_bal_index();
1251 }
1252
1253 let exec_span = debug_span!(target: "engine::tree", "execution").entered();
1255 let mut transactions = transactions.into_iter();
1256 let mut last_sent_len = 0usize;
1261 loop {
1262 let wait_start = Instant::now();
1265 let Some(tx_result) = transactions.next() else { break };
1266 self.metrics.record_transaction_wait(wait_start.elapsed());
1267
1268 let tx = tx_result.map_err(BlockValidationError::other)?;
1269 let tx_signer = *<Tx as alloy_evm::RecoveredTx<InnerTx>>::signer(&tx);
1270
1271 senders.push(tx_signer);
1272
1273 let _enter = tracing::enabled!(target: "engine::tree", Level::TRACE).then(|| {
1274 tracing::trace_span!(
1275 target: "engine::tree",
1276 "execute tx",
1277 tx_index = senders.len() - 1,
1278 )
1279 .entered()
1280 });
1281 if tracing::enabled!(target: "engine::tree", Level::TRACE) {
1282 trace!(target: "engine::tree", "Executing transaction");
1283 }
1284
1285 let tx_start = Instant::now();
1286 executor.execute_transaction(tx)?;
1287 self.metrics.record_transaction_execution(tx_start.elapsed());
1288
1289 executed_tx_index.store(senders.len(), Ordering::Relaxed);
1291
1292 let current_len = executor.receipts().len();
1293 if current_len > last_sent_len {
1294 last_sent_len = current_len;
1295 if let Some(receipt) = executor.receipts().last() {
1297 let tx_index = current_len - 1;
1298 let _ = receipt_tx.send(IndexedReceipt::new(tx_index, receipt.clone()));
1299 }
1300 }
1301 if has_bal {
1303 executor.evm_mut().db_mut().bump_bal_index();
1304 }
1305 }
1306
1307 drop(exec_span);
1308
1309 Ok((executor, senders))
1310 }
1311
1312 #[instrument(level = "debug", target = "engine::tree::payload_validator", skip_all)]
1324 fn validate_post_execution<T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>>(
1325 &mut self,
1326 block: &RecoveredBlock<N::Block>,
1327 parent_block: &SealedHeader<N::BlockHeader>,
1328 output: &BlockExecutionOutput<N::Receipt>,
1329 ctx: &mut TreeCtx<'_, N>,
1330 receipt_root_bloom: Option<ReceiptRootBloom>,
1331 built_bal: Option<BlockAccessList>,
1332 ) -> Result<(), InsertBlockErrorKind>
1333 where
1334 V: PayloadValidator<T, Block = N::Block>,
1335 {
1336 let start = Instant::now();
1337
1338 trace!(target: "engine::tree::payload_validator", block=?block.num_hash(), "Validating block consensus");
1339
1340 let _enter =
1342 debug_span!(target: "engine::tree::payload_validator", "validate_block_post_execution")
1343 .entered();
1344 let built_bal = built_bal.map(Bal::from);
1345 let block_access_list_hash =
1346 built_bal.as_ref().map(|bal| bal.compute_hash_with_buf(&mut self.bal_hash_buf));
1347
1348 let validation_result = built_bal
1349 .as_ref()
1350 .map(|bal| bal.validate_gas_limit(block.gas_limit()).map_err(ConsensusError::from))
1351 .transpose()
1352 .and_then(|_| {
1353 self.consensus.validate_block_post_execution(
1354 block,
1355 output,
1356 receipt_root_bloom,
1357 block_access_list_hash,
1358 )
1359 });
1360
1361 if let Err(err) = validation_result {
1362 self.on_invalid_block(parent_block, block, output, None, ctx.state_mut());
1364 return Err(err.into())
1365 }
1366 drop(_enter);
1367
1368 self.metrics
1370 .block_validation
1371 .post_execution_validation_duration
1372 .record(start.elapsed().as_secs_f64());
1373
1374 Ok(())
1375 }
1376
1377 #[instrument(
1382 level = "debug",
1383 target = "engine::tree::payload_validator",
1384 skip_all,
1385 fields(
1386 has_hint_stream = hint_stream.is_some(),
1387 has_hashed_update_stream = hashed_update_stream.is_some(),
1388 parallel_bal_execution
1389 )
1390 )]
1391 fn spawn_payload_processor<T: ExecutableTxIterator<Evm>>(
1392 &self,
1393 env: ExecutionEnv<Evm>,
1394 txs: T,
1395 state_provider_factory: OverlayStateProviderFactory<P, N>,
1396 hint_stream: Option<StateRootHintStream>,
1397 hashed_update_stream: Option<StateRootUpdateStream>,
1398 parallel_bal_execution: bool,
1399 ) -> Result<
1400 PayloadHandle<
1401 impl ExecutableTxFor<Evm> + use<N, P, Evm, V, T>,
1402 impl core::error::Error + Send + Sync + 'static + use<N, P, Evm, V, T>,
1403 N::Receipt,
1404 >,
1405 InsertBlockErrorKind,
1406 > {
1407 let start = Instant::now();
1408 let handle = self.payload_processor.spawn_with_state_root_streams(
1409 env,
1410 txs,
1411 state_provider_factory,
1412 hint_stream,
1413 hashed_update_stream,
1414 parallel_bal_execution,
1415 );
1416
1417 self.metrics.block_validation.spawn_payload_processor.record(start.elapsed().as_secs_f64());
1418
1419 Ok(handle)
1420 }
1421
1422 fn overlay_state_provider_factory(
1426 &self,
1427 hash: B256,
1428 state: &EngineApiTreeState<N>,
1429 ) -> ProviderResult<Option<OverlayStateProviderFactory<P, N>>> {
1430 if !state.tree_state.contains_hash(&hash) && self.provider.header(hash)?.is_none() {
1431 debug!(target: "engine::tree::payload_validator", %hash, "no canonical state found for block");
1432 return Ok(None)
1433 }
1434
1435 Ok(Some(OverlayStateProviderFactory::new(
1436 self.provider.clone(),
1437 state.tree_state.overlay_manager.overlay_builder(hash),
1438 )))
1439 }
1440
1441 fn on_invalid_block(
1443 &self,
1444 parent_header: &SealedHeader<N::BlockHeader>,
1445 block: &RecoveredBlock<N::Block>,
1446 output: &BlockExecutionOutput<N::Receipt>,
1447 trie_updates: Option<(&TrieUpdates, B256)>,
1448 state: &mut EngineApiTreeState<N>,
1449 ) {
1450 if state.invalid_headers.get(&block.hash()).is_some() {
1451 return
1453 }
1454 self.invalid_block_hook.on_invalid_block(parent_header, block, output, trie_updates);
1455 }
1456
1457 fn payload_state_root_handle_for(
1460 &self,
1461 parent_hash: B256,
1462 parent_header: &N::BlockHeader,
1463 timestamp: u64,
1464 state: &mut EngineApiTreeState<N>,
1465 ) -> Option<PayloadStateRootHandle> {
1466 let state_provider_factory = match self.overlay_state_provider_factory(parent_hash, state) {
1467 Ok(Some(state_provider_factory)) => state_provider_factory,
1468 Ok(None) => return None,
1469 Err(err) => {
1470 warn!(
1471 target: "engine::tree::payload_validator",
1472 %err,
1473 %parent_hash,
1474 "failed to prepare payload-builder state-root provider"
1475 );
1476 return None
1477 }
1478 };
1479 match self.state_root_strategy.prepare_payload_builder(PayloadStateRootJobContext::new(
1480 &self.runtime,
1481 &self.overlay_manager,
1482 parent_hash,
1483 parent_header,
1484 timestamp,
1485 state,
1486 state_provider_factory,
1487 &self.config,
1488 )) {
1489 Ok(handle) => handle,
1490 Err(err) => {
1491 warn!(
1492 target: "engine::tree::payload_validator",
1493 %err,
1494 %parent_hash,
1495 "failed to prepare payload-builder state-root job"
1496 );
1497 None
1498 }
1499 }
1500 }
1501
1502 fn spawn_deferred_trie_task(
1515 &self,
1516 block: Arc<RecoveredBlock<N::Block>>,
1517 execution_outcome: Arc<BlockExecutionOutput<N::Receipt>>,
1518 hashed_state: LazyHashedPostState,
1519 trie_output: Arc<TrieUpdates>,
1520 ) -> ExecutedBlock<N> {
1521 let hashed_state = match hashed_state.try_into_inner() {
1525 Ok(state) => state,
1526 Err(handle) => handle.get().clone(),
1527 };
1528 let (deferred_trie_data, deferred_trie_task) =
1529 LazyTrieData::pending(hashed_state, trie_output);
1530 let block_validation_metrics = self.metrics.block_validation.clone();
1531
1532 let block_number = block.number();
1534
1535 let compute_trie_input_task = move || {
1537 let _span = debug_span!(
1538 target: "engine::tree::payload_validator",
1539 "compute_trie_input_task",
1540 block_number
1541 )
1542 .entered();
1543
1544 let compute_start = Instant::now();
1545 let computed = deferred_trie_task.compute_and_publish();
1546 block_validation_metrics
1547 .deferred_trie_compute_duration
1548 .record(compute_start.elapsed().as_secs_f64());
1549
1550 block_validation_metrics
1552 .hashed_post_state_size
1553 .record(computed.sorted.hashed_state.total_len() as f64);
1554 block_validation_metrics
1555 .trie_updates_sorted_size
1556 .record(computed.sorted.trie_updates.total_len() as f64);
1557 };
1558
1559 self.runtime.spawn_blocking_named(DEFERRED_TRIE_WORKER_NAME, compute_trie_input_task);
1561
1562 ExecutedBlock::with_deferred_trie_data(block, execution_outcome, deferred_trie_data)
1563 }
1564
1565 fn calculate_timing_stats(
1566 &self,
1567 block: &RecoveredBlock<N::Block>,
1568 provider_stats: Arc<StateProviderStats>,
1569 cache_stats: Option<Arc<CacheStats>>,
1570 output: &BlockExecutionOutput<N::Receipt>,
1571 execution_duration: Duration,
1572 state_hash_duration: Duration,
1573 ) -> Box<ExecutionTimingStats> {
1574 let accounts_read = provider_stats.total_account_fetches();
1575 let storage_read = provider_stats.total_storage_fetches();
1576 let code_read = provider_stats.total_code_fetches();
1577 let code_bytes_read = provider_stats.total_code_fetched_bytes();
1578
1579 let accounts_changed = output.state.state.len();
1581 let accounts_deleted =
1582 output.state.state.values().filter(|acc| acc.was_destroyed()).count();
1583 let storage_slots_changed =
1584 output.state.state.values().map(|account| account.storage.len()).sum::<usize>();
1585 let storage_slots_deleted = output
1586 .state
1587 .state
1588 .values()
1589 .flat_map(|account| account.storage.values())
1590 .filter(|slot| {
1591 slot.present_value.is_zero() && !slot.previous_or_original_value.is_zero()
1592 })
1593 .count();
1594
1595 let is_new_deployment = |acc: &BundleAccount| -> bool {
1597 let has_code_now = acc.info.as_ref().is_some_and(|info| info.code_hash != KECCAK_EMPTY);
1598 let had_no_code_before = acc
1599 .original_info
1600 .as_ref()
1601 .map(|info| info.code_hash == KECCAK_EMPTY)
1602 .unwrap_or(true);
1603 has_code_now && had_no_code_before
1604 };
1605
1606 let bytecodes_changed =
1607 output.state.state.values().filter(|acc| is_new_deployment(acc)).count();
1608
1609 let unique_new_code_hashes: B256Set = output
1611 .state
1612 .state
1613 .values()
1614 .filter(|acc| is_new_deployment(acc))
1615 .filter_map(|acc| acc.info.as_ref().map(|info| info.code_hash))
1616 .collect();
1617 let code_bytes_written: usize = unique_new_code_hashes
1618 .iter()
1619 .filter_map(|hash| {
1620 output.state.contracts.get(hash).map(|bytecode| bytecode.original_bytes().len())
1621 })
1622 .sum();
1623
1624 let state_read_duration = provider_stats.total_account_fetch_latency() +
1626 provider_stats.total_storage_fetch_latency() +
1627 provider_stats.total_code_fetch_latency();
1628
1629 let eip7702_delegations_set =
1632 output.state.contracts.values().filter(|bytecode| bytecode.is_eip7702()).count();
1633 let eip7702_delegations_cleared = output
1638 .state
1639 .state
1640 .values()
1641 .filter(|acc| {
1642 let original_was_eip7702 = acc
1644 .original_info
1645 .as_ref()
1646 .and_then(|info| info.code.as_ref())
1647 .map(|bytecode| bytecode.is_eip7702())
1648 .unwrap_or(false);
1649
1650 let code_now_empty =
1652 acc.info.as_ref().map(|info| info.code_hash == KECCAK_EMPTY).unwrap_or(false);
1653
1654 original_was_eip7702 && code_now_empty
1655 })
1656 .count();
1657
1658 let (account_cache_hits, account_cache_misses) = cache_stats
1660 .as_ref()
1661 .map(|s| (s.account_hits(), s.account_misses()))
1662 .unwrap_or_default();
1663 let (storage_cache_hits, storage_cache_misses) = cache_stats
1664 .as_ref()
1665 .map(|s| (s.storage_hits(), s.storage_misses()))
1666 .unwrap_or_default();
1667 let (code_cache_hits, code_cache_misses) =
1668 cache_stats.as_ref().map(|s| (s.code_hits(), s.code_misses())).unwrap_or_default();
1669 let (txpool_snapshot_account_hits, txpool_snapshot_account_misses) = cache_stats
1670 .as_ref()
1671 .map(|s| (s.txpool_snapshot_account_hits(), s.txpool_snapshot_account_misses()))
1672 .unwrap_or_default();
1673 let (txpool_snapshot_storage_hits, txpool_snapshot_storage_misses) = cache_stats
1674 .as_ref()
1675 .map(|s| (s.txpool_snapshot_storage_hits(), s.txpool_snapshot_storage_misses()))
1676 .unwrap_or_default();
1677 let (txpool_snapshot_code_hits, txpool_snapshot_code_misses) = cache_stats
1678 .as_ref()
1679 .map(|s| (s.txpool_snapshot_code_hits(), s.txpool_snapshot_code_misses()))
1680 .unwrap_or_default();
1681
1682 Box::new(ExecutionTimingStats {
1684 block_number: block.number(),
1685 block_hash: block.hash(),
1686 gas_used: output.result.gas_used,
1687 tx_count: block.transaction_count(),
1688 execution_duration,
1689 state_read_duration,
1690 state_hash_duration,
1691 accounts_read,
1692 storage_read,
1693 code_read,
1694 code_bytes_read,
1695 accounts_changed,
1696 accounts_deleted,
1697 storage_slots_changed,
1698 storage_slots_deleted,
1699 bytecodes_changed,
1700 code_bytes_written,
1701 eip7702_delegations_set,
1702 eip7702_delegations_cleared,
1703 account_cache_hits,
1704 account_cache_misses,
1705 storage_cache_hits,
1706 storage_cache_misses,
1707 code_cache_hits,
1708 code_cache_misses,
1709 txpool_snapshot_account_hits,
1710 txpool_snapshot_account_misses,
1711 txpool_snapshot_storage_hits,
1712 txpool_snapshot_storage_misses,
1713 txpool_snapshot_code_hits,
1714 txpool_snapshot_code_misses,
1715 })
1716 }
1717}
1718
1719pub trait EngineValidator<
1723 Types: PayloadTypes,
1724 N: NodePrimitives = <<Types as PayloadTypes>::BuiltPayload as BuiltPayload>::Primitives,
1725>: Send + Sync + 'static
1726{
1727 fn validate_payload_attributes_against_header(
1737 &self,
1738 attr: &Types::PayloadAttributes,
1739 header: &N::BlockHeader,
1740 ) -> Result<(), InvalidPayloadAttributesError>;
1741
1742 fn convert_payload_to_block(
1751 &self,
1752 payload: Types::ExecutionData,
1753 ) -> Result<SealedBlock<N::Block>, NewPayloadError>;
1754
1755 fn validate_payload(
1757 &mut self,
1758 payload: Types::ExecutionData,
1759 ctx: TreeCtx<'_, N>,
1760 ) -> ValidationOutcome<N>;
1761
1762 fn validate_block(
1764 &mut self,
1765 block: SealedBlockWithAccessList<N::Block>,
1766 ctx: TreeCtx<'_, N>,
1767 ) -> ValidationOutcome<N>;
1768
1769 fn on_inserted_executed_block(
1774 &self,
1775 block: BuiltPayloadExecutedBlock<N>,
1776 ) -> ProviderResult<ExecutedBlock<N>>;
1777
1778 fn on_canonical_head_changed(&self, _hash: B256, _state: &EngineApiTreeState<N>) {}
1782
1783 fn payload_builder_resources(
1787 &self,
1788 parent_hash: B256,
1789 parent_header: &N::BlockHeader,
1790 timestamp: u64,
1791 state: &mut EngineApiTreeState<N>,
1792 ) -> PayloadBuilderResources;
1793}
1794
1795impl<N, Types, P, Evm, V> EngineValidator<Types> for BasicEngineValidator<P, Evm, V>
1796where
1797 P: DatabaseProviderFactory<
1798 Provider: BlockReader
1799 + BlockHashReader
1800 + StageCheckpointReader
1801 + PruneCheckpointReader
1802 + ChangeSetReader
1803 + StorageChangeSetReader
1804 + StorageSettingsCache
1805 + HistoryReader
1806 + 'static,
1807 > + BlockReader<Header = N::BlockHeader>
1808 + StateProviderFactory
1809 + StateReader
1810 + ChangeSetReader
1811 + Clone
1812 + 'static,
1813 OverlayStateProviderFactory<P, N>: DatabaseProviderROFactory<
1814 Provider: TrieCursorFactory
1815 + HashedCursorFactory
1816 + HashedPostStateProvider
1817 + StateRootProvider
1818 + StateProvider
1819 + Send,
1820 > + Clone
1821 + 'static,
1822 N: NodePrimitives,
1823 V: PayloadValidator<Types, Block = N::Block> + Clone,
1824 Evm: ConfigureEngineEvm<Types::ExecutionData, Primitives = N> + 'static,
1825 Types: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>,
1826{
1827 fn validate_payload_attributes_against_header(
1828 &self,
1829 attr: &Types::PayloadAttributes,
1830 header: &N::BlockHeader,
1831 ) -> Result<(), InvalidPayloadAttributesError> {
1832 self.validator.validate_payload_attributes_against_header(attr, header)
1833 }
1834
1835 fn convert_payload_to_block(
1836 &self,
1837 payload: Types::ExecutionData,
1838 ) -> Result<SealedBlock<N::Block>, NewPayloadError> {
1839 let block = self.validator.convert_payload_to_block(payload)?;
1840 Ok(block)
1841 }
1842
1843 fn validate_payload(
1844 &mut self,
1845 payload: Types::ExecutionData,
1846 ctx: TreeCtx<'_, N>,
1847 ) -> ValidationOutcome<N> {
1848 self.validate_block_with_state(BlockOrPayload::Payload(payload), ctx)
1849 }
1850
1851 fn validate_block(
1852 &mut self,
1853 block: SealedBlockWithAccessList<N::Block>,
1854 ctx: TreeCtx<'_, N>,
1855 ) -> ValidationOutcome<N> {
1856 self.validate_block_with_state(BlockOrPayload::Block(block), ctx)
1857 }
1858
1859 fn on_inserted_executed_block(
1860 &self,
1861 block: BuiltPayloadExecutedBlock<N>,
1862 ) -> ProviderResult<ExecutedBlock<N>> {
1863 self.payload_processor.on_inserted_executed_block(
1864 block.recovered_block.block_with_parent(),
1865 &block.execution_output.state,
1866 );
1867
1868 Ok(self.spawn_deferred_trie_task(
1869 block.recovered_block,
1870 block.execution_output,
1871 LazyHashedPostState::ready(block.hashed_state),
1872 block.trie_updates,
1873 ))
1874 }
1875
1876 fn on_canonical_head_changed(&self, hash: B256, state: &EngineApiTreeState<N>) {
1877 let Some(txpool_prewarm) = self.txpool_prewarm.as_ref() else { return };
1878
1879 let parent = match self.sealed_header_by_hash(hash, state) {
1882 Ok(Some(header)) => header,
1883 Ok(None) => return,
1884 Err(err) => {
1885 trace!(
1886 target: "engine::tree::txpool_prewarm",
1887 %err,
1888 block_hash = ?hash,
1889 "failed to fetch canonical header for txpool prewarming"
1890 );
1891 return
1892 }
1893 };
1894 let evm_env = match self.evm_config.evm_env(parent.header()) {
1898 Ok(evm_env) => evm_env,
1899 Err(err) => {
1900 trace!(
1901 target: "engine::tree::txpool_prewarm",
1902 %err,
1903 block_hash = ?parent.hash(),
1904 "failed to derive canonical txpool prewarming environment"
1905 );
1906 return
1907 }
1908 };
1909
1910 let state_provider_factory = match self.overlay_state_provider_factory(parent.hash(), state)
1911 {
1912 Ok(Some(state_provider_factory)) => state_provider_factory,
1913 Ok(None) => return,
1914 Err(err) => {
1915 trace!(
1916 target: "engine::tree::txpool_prewarm",
1917 %err,
1918 block_hash = ?parent.hash(),
1919 "failed to derive canonical txpool prewarming provider"
1920 );
1921 return
1922 }
1923 };
1924 txpool_prewarm.start(parent.hash(), evm_env, state_provider_factory)
1925 }
1926
1927 fn payload_builder_resources(
1928 &self,
1929 parent_hash: B256,
1930 parent_header: &N::BlockHeader,
1931 timestamp: u64,
1932 state: &mut EngineApiTreeState<N>,
1933 ) -> PayloadBuilderResources {
1934 let execution_cache = self
1935 .config
1936 .share_execution_cache_with_payload_builder()
1937 .then(|| self.payload_processor.cache_for(parent_hash));
1938 let state_root_handle =
1939 self.payload_state_root_handle_for(parent_hash, parent_header, timestamp, state);
1940 let mut resources = PayloadBuilderResources::new(execution_cache, state_root_handle)
1941 .with_lease(PayloadBuilderLease::new(JitPauseGuard::new(&self.evm_config)));
1942 if let Some(txpool_prewarm) = self.txpool_prewarm.as_ref() {
1946 let txpool_lease = PayloadBuilderLease::new(txpool_prewarm.pause());
1947 resources = resources.with_lease(txpool_lease);
1948 }
1949 resources
1950 }
1951}
1952
1953impl<P, Evm, V> WaitForCaches for BasicEngineValidator<P, Evm, V>
1954where
1955 Evm: ConfigureEvm,
1956{
1957 fn wait_for_caches(&self) -> CacheWaitDurations {
1958 debug!(target: "engine::tree::payload_validator", "Waiting for execution cache and sparse trie locks");
1959
1960 let execution_cache = self.payload_processor.execution_cache();
1961 let overlay_manager = self.overlay_manager.clone();
1962 let (execution_tx, execution_rx) = std::sync::mpsc::channel();
1963 let (sparse_trie_tx, sparse_trie_rx) = std::sync::mpsc::channel();
1964
1965 self.runtime.spawn_blocking_named("wait-exec-cache", move || {
1966 let _ = execution_tx.send(execution_cache.wait_for_availability());
1967 });
1968 self.runtime.spawn_blocking_named("wait-sparse-tri", move || {
1969 let _ = sparse_trie_tx.send(overlay_manager.wait_for_sparse_trie_availability());
1970 });
1971
1972 let execution_cache =
1973 execution_rx.recv().expect("execution cache wait task failed to send result");
1974 let sparse_trie =
1975 sparse_trie_rx.recv().expect("sparse trie wait task failed to send result");
1976 debug!(
1977 target: "engine::tree::payload_validator",
1978 ?execution_cache,
1979 ?sparse_trie,
1980 "Execution cache and sparse trie locks acquired"
1981 );
1982 CacheWaitDurations { execution_cache, sparse_trie }
1983 }
1984}
1985
1986#[derive(Debug, Clone)]
1988pub enum BlockOrPayload<T: PayloadTypes> {
1989 Payload(T::ExecutionData),
1991 Block(SealedBlockWithAccessList<BlockTy<<T::BuiltPayload as BuiltPayload>::Primitives>>),
1993}
1994
1995impl<T: PayloadTypes> BlockOrPayload<T> {
1996 pub fn hash(&self) -> B256 {
1998 match self {
1999 Self::Payload(payload) => payload.block_hash(),
2000 Self::Block(block) => block.hash(),
2001 }
2002 }
2003
2004 pub fn num_hash(&self) -> NumHash {
2006 match self {
2007 Self::Payload(payload) => payload.num_hash(),
2008 Self::Block(block) => block.num_hash(),
2009 }
2010 }
2011
2012 pub fn parent_hash(&self) -> B256 {
2014 match self {
2015 Self::Payload(payload) => payload.parent_hash(),
2016 Self::Block(block) => block.parent_hash(),
2017 }
2018 }
2019
2020 pub fn block_with_parent(&self) -> BlockWithParent {
2022 match self {
2023 Self::Payload(payload) => payload.block_with_parent(),
2024 Self::Block(block) => block.block_with_parent(),
2025 }
2026 }
2027
2028 pub const fn type_name(&self) -> &'static str {
2030 match self {
2031 Self::Payload(_) => "payload",
2032 Self::Block(_) => "block",
2033 }
2034 }
2035
2036 pub const fn is_payload(&self) -> bool {
2038 matches!(self, Self::Payload(_))
2039 }
2040
2041 pub const fn is_block(&self) -> bool {
2043 matches!(self, Self::Block(_))
2044 }
2045
2046 pub fn try_decoded_access_list(&self) -> Result<Option<DecodedBal>, alloy_rlp::Error> {
2048 match self {
2049 Self::Payload(payload) => payload
2050 .block_access_list()
2051 .map(|block_access_list| DecodedBal::from_rlp_bytes(block_access_list.clone()))
2052 .transpose(),
2053 Self::Block(block) => block.data().clone().map(DecodedBal::from_raw_bal).transpose(),
2054 }
2055 }
2056
2057 pub fn has_block_access_list(&self) -> bool {
2059 match self {
2060 Self::Payload(payload) => payload.block_access_list().is_some(),
2061 Self::Block(block) => block.block_access_list_hash().is_some(),
2062 }
2063 }
2064
2065 pub fn transaction_count(&self) -> usize
2067 where
2068 T::ExecutionData: ExecutionPayload,
2069 {
2070 match self {
2071 Self::Payload(payload) => payload.transaction_count(),
2072 Self::Block(block) => block.transaction_count(),
2073 }
2074 }
2075
2076 pub fn withdrawals(&self) -> Option<&[Withdrawal]>
2078 where
2079 T::ExecutionData: ExecutionPayload,
2080 {
2081 match self {
2082 Self::Payload(payload) => payload.withdrawals().map(|w| w.as_slice()),
2083 Self::Block(block) => block.body().withdrawals().map(|w| w.as_slice()),
2084 }
2085 }
2086
2087 pub fn gas_used(&self) -> u64
2089 where
2090 T::ExecutionData: ExecutionPayload,
2091 {
2092 match self {
2093 Self::Payload(payload) => payload.gas_used(),
2094 Self::Block(block) => block.gas_used(),
2095 }
2096 }
2097
2098 pub fn gas_limit(&self) -> u64
2100 where
2101 T::ExecutionData: ExecutionPayload,
2102 {
2103 match self {
2104 Self::Payload(payload) => payload.gas_limit(),
2105 Self::Block(block) => block.gas_limit(),
2106 }
2107 }
2108}