reth_engine_tree/tree/

mod.rs

use crate::{
    backfill::{BackfillAction, BackfillSyncState},
    chain::FromOrchestrator,
    engine::{DownloadRequest, EngineApiEvent, EngineApiKind, EngineApiRequest, FromEngine},
    persistence::PersistenceHandle,
    tree::{
        cached_state::CachedStateProvider, executor::WorkloadExecutor, metrics::EngineApiMetrics,
    },
};
use alloy_consensus::BlockHeader;
use alloy_eips::{merge::EPOCH_SLOTS, BlockNumHash};
use alloy_primitives::{
    map::{HashMap, HashSet},
    BlockNumber, B256, U256,
};
use alloy_rpc_types_engine::{
    ForkchoiceState, PayloadStatus, PayloadStatusEnum, PayloadValidationError,
};
use error::{InsertBlockError, InsertBlockErrorKind, InsertBlockFatalError};
use payload_processor::sparse_trie::StateRootComputeOutcome;
use persistence_state::CurrentPersistenceAction;
use reth_chain_state::{
    CanonicalInMemoryState, ExecutedBlock, ExecutedBlockWithTrieUpdates,
    MemoryOverlayStateProvider, NewCanonicalChain,
};
use reth_consensus::{Consensus, FullConsensus};
pub use reth_engine_primitives::InvalidBlockHook;
use reth_engine_primitives::{
    BeaconConsensusEngineEvent, BeaconEngineMessage, BeaconOnNewPayloadError, EngineTypes,
    EngineValidator, ExecutionPayload, ForkchoiceStateTracker, OnForkChoiceUpdated,
};
use reth_errors::{ConsensusError, ProviderResult};
use reth_ethereum_primitives::EthPrimitives;
use reth_evm::{execute::BlockExecutorProvider, ConfigureEvm};
use reth_payload_builder::PayloadBuilderHandle;
use reth_payload_primitives::{EngineApiMessageVersion, PayloadBuilderAttributes};
use reth_primitives_traits::{
    Block, GotExpected, NodePrimitives, RecoveredBlock, SealedBlock, SealedHeader,
};
use reth_provider::{
    providers::ConsistentDbView, BlockNumReader, BlockReader, DBProvider, DatabaseProviderFactory,
    ExecutionOutcome, HashedPostStateProvider, ProviderError, StateCommitmentProvider,
    StateProviderBox, StateProviderFactory, StateReader, StateRootProvider, TransactionVariant,
};
use reth_revm::database::StateProviderDatabase;
use reth_stages_api::ControlFlow;
use reth_trie::{updates::TrieUpdates, HashedPostState, TrieInput};
use reth_trie_db::{DatabaseHashedPostState, StateCommitment};
use reth_trie_parallel::root::{ParallelStateRoot, ParallelStateRootError};
use std::{
    cmp::Ordering,
    collections::{btree_map, hash_map, BTreeMap, VecDeque},
    fmt::Debug,
    ops::Bound,
    sync::{
        mpsc::{Receiver, RecvError, RecvTimeoutError, Sender},
        Arc,
    },
    time::Instant,
};
use tokio::sync::{
    mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender},
    oneshot::{self, error::TryRecvError},
};
use tracing::*;

mod block_buffer;
mod cached_state;
pub mod error;
mod invalid_block_hook;
mod invalid_headers;
mod metrics;
mod payload_processor;
mod persistence_state;
// TODO(alexey): compare trie updates in `insert_block_inner`
#[allow(unused)]
mod trie_updates;

use crate::tree::error::AdvancePersistenceError;
pub use block_buffer::BlockBuffer;
pub use invalid_block_hook::{InvalidBlockHooks, NoopInvalidBlockHook};
pub use invalid_headers::InvalidHeaderCache;
pub use payload_processor::*;
pub use persistence_state::PersistenceState;
pub use reth_engine_primitives::TreeConfig;
use reth_evm::execute::BlockExecutionOutput;

/// The largest gap for which the tree will be used for sync. See docs for `pipeline_run_threshold`
/// for more information.
///
/// This is the default threshold, the distance to the head that the tree will be used for sync.
/// If the distance exceeds this threshold, the pipeline will be used for sync.
pub(crate) const MIN_BLOCKS_FOR_PIPELINE_RUN: u64 = EPOCH_SLOTS;

/// Keeps track of the state of the tree.
///
/// ## Invariants
///
/// - This only stores blocks that are connected to the canonical chain.
/// - All executed blocks are valid and have been executed.
#[derive(Debug, Default)]
pub struct TreeState<N: NodePrimitives = EthPrimitives> {
    /// __All__ unique executed blocks by block hash that are connected to the canonical chain.
    ///
    /// This includes blocks of all forks.
    blocks_by_hash: HashMap<B256, ExecutedBlockWithTrieUpdates<N>>,
    /// Executed blocks grouped by their respective block number.
    ///
    /// This maps unique block number to all known blocks for that height.
    ///
    /// Note: there can be multiple blocks at the same height due to forks.
    blocks_by_number: BTreeMap<BlockNumber, Vec<ExecutedBlockWithTrieUpdates<N>>>,
    /// Map of any parent block hash to its children.
    parent_to_child: HashMap<B256, HashSet<B256>>,
    /// Map of hash to trie updates for canonical blocks that are persisted but not finalized.
    ///
    /// Contains the block number for easy removal.
    persisted_trie_updates: HashMap<B256, (BlockNumber, Arc<TrieUpdates>)>,
    /// Currently tracked canonical head of the chain.
    current_canonical_head: BlockNumHash,
}

impl<N: NodePrimitives> TreeState<N> {
    /// Returns a new, empty tree state that points to the given canonical head.
    fn new(current_canonical_head: BlockNumHash) -> Self {
        Self {
            blocks_by_hash: HashMap::default(),
            blocks_by_number: BTreeMap::new(),
            current_canonical_head,
            parent_to_child: HashMap::default(),
            persisted_trie_updates: HashMap::default(),
        }
    }

    /// Resets the state and points to the given canonical head.
    fn reset(&mut self, current_canonical_head: BlockNumHash) {
        *self = Self::new(current_canonical_head);
    }

    /// Returns the number of executed blocks stored.
    fn block_count(&self) -> usize {
        self.blocks_by_hash.len()
    }

    /// Returns the [`ExecutedBlockWithTrieUpdates`] by hash.
    fn executed_block_by_hash(&self, hash: B256) -> Option<&ExecutedBlockWithTrieUpdates<N>> {
        self.blocks_by_hash.get(&hash)
    }

    /// Returns the block by hash.
    fn block_by_hash(&self, hash: B256) -> Option<Arc<SealedBlock<N::Block>>> {
        self.blocks_by_hash.get(&hash).map(|b| Arc::new(b.recovered_block().sealed_block().clone()))
    }

    /// Returns all available blocks for the given hash that lead back to the canonical chain, from
    /// newest to oldest. And the parent hash of the oldest block that is missing from the buffer.
    ///
    /// Returns `None` if the block for the given hash is not found.
    fn blocks_by_hash(&self, hash: B256) -> Option<(B256, Vec<ExecutedBlockWithTrieUpdates<N>>)> {
        let block = self.blocks_by_hash.get(&hash).cloned()?;
        let mut parent_hash = block.recovered_block().parent_hash();
        let mut blocks = vec![block];
        while let Some(executed) = self.blocks_by_hash.get(&parent_hash) {
            parent_hash = executed.recovered_block().parent_hash();
            blocks.push(executed.clone());
        }

        Some((parent_hash, blocks))
    }

    /// Insert executed block into the state.
    fn insert_executed(&mut self, executed: ExecutedBlockWithTrieUpdates<N>) {
        let hash = executed.recovered_block().hash();
        let parent_hash = executed.recovered_block().parent_hash();
        let block_number = executed.recovered_block().number();

        if self.blocks_by_hash.contains_key(&hash) {
            return;
        }

        self.blocks_by_hash.insert(hash, executed.clone());

        self.blocks_by_number.entry(block_number).or_default().push(executed);

        self.parent_to_child.entry(parent_hash).or_default().insert(hash);

        for children in self.parent_to_child.values_mut() {
            children.retain(|child| self.blocks_by_hash.contains_key(child));
        }
    }

    /// Remove single executed block by its hash.
    ///
    /// ## Returns
    ///
    /// The removed block and the block hashes of its children.
    fn remove_by_hash(
        &mut self,
        hash: B256,
    ) -> Option<(ExecutedBlockWithTrieUpdates<N>, HashSet<B256>)> {
        let executed = self.blocks_by_hash.remove(&hash)?;

        // Remove this block from collection of children of its parent block.
        let parent_entry = self.parent_to_child.entry(executed.recovered_block().parent_hash());
        if let hash_map::Entry::Occupied(mut entry) = parent_entry {
            entry.get_mut().remove(&hash);

            if entry.get().is_empty() {
                entry.remove();
            }
        }

        // Remove point to children of this block.
        let children = self.parent_to_child.remove(&hash).unwrap_or_default();

        // Remove this block from `blocks_by_number`.
        let block_number_entry = self.blocks_by_number.entry(executed.recovered_block().number());
        if let btree_map::Entry::Occupied(mut entry) = block_number_entry {
            // We have to find the index of the block since it exists in a vec
            if let Some(index) = entry.get().iter().position(|b| b.recovered_block().hash() == hash)
            {
                entry.get_mut().swap_remove(index);

                // If there are no blocks left then remove the entry for this block
                if entry.get().is_empty() {
                    entry.remove();
                }
            }
        }

        Some((executed, children))
    }

    /// Returns whether or not the hash is part of the canonical chain.
    pub(crate) fn is_canonical(&self, hash: B256) -> bool {
        let mut current_block = self.current_canonical_head.hash;
        if current_block == hash {
            return true
        }

        while let Some(executed) = self.blocks_by_hash.get(&current_block) {
            current_block = executed.recovered_block().parent_hash();
            if current_block == hash {
                return true
            }
        }

        false
    }

    /// Removes canonical blocks below the upper bound, only if the last persisted hash is
    /// part of the canonical chain.
    pub(crate) fn remove_canonical_until(
        &mut self,
        upper_bound: BlockNumber,
        last_persisted_hash: B256,
    ) {
        debug!(target: "engine::tree", ?upper_bound, ?last_persisted_hash, "Removing canonical blocks from the tree");

        // If the last persisted hash is not canonical, then we don't want to remove any canonical
        // blocks yet.
        if !self.is_canonical(last_persisted_hash) {
            return
        }

        // First, let's walk back the canonical chain and remove canonical blocks lower than the
        // upper bound
        let mut current_block = self.current_canonical_head.hash;
        while let Some(executed) = self.blocks_by_hash.get(&current_block) {
            current_block = executed.recovered_block().parent_hash();
            if executed.recovered_block().number() <= upper_bound {
                debug!(target: "engine::tree", num_hash=?executed.recovered_block().num_hash(), "Attempting to remove block walking back from the head");
                if let Some((removed, _)) = self.remove_by_hash(executed.recovered_block().hash()) {
                    debug!(target: "engine::tree", num_hash=?removed.recovered_block().num_hash(), "Removed block walking back from the head");
                    // finally, move the trie updates
                    self.persisted_trie_updates.insert(
                        removed.recovered_block().hash(),
                        (removed.recovered_block().number(), removed.trie),
                    );
                }
            }
        }
        debug!(target: "engine::tree", ?upper_bound, ?last_persisted_hash, "Removed canonical blocks from the tree");
    }

    /// Removes all blocks that are below the finalized block, as well as removing non-canonical
    /// sidechains that fork from below the finalized block.
    pub(crate) fn prune_finalized_sidechains(&mut self, finalized_num_hash: BlockNumHash) {
        let BlockNumHash { number: finalized_num, hash: finalized_hash } = finalized_num_hash;

        // We remove disconnected sidechains in three steps:
        // * first, remove everything with a block number __below__ the finalized block.
        // * next, we populate a vec with parents __at__ the finalized block.
        // * finally, we iterate through the vec, removing children until the vec is empty
        // (BFS).

        // We _exclude_ the finalized block because we will be dealing with the blocks __at__
        // the finalized block later.
        let blocks_to_remove = self
            .blocks_by_number
            .range((Bound::Unbounded, Bound::Excluded(finalized_num)))
            .flat_map(|(_, blocks)| blocks.iter().map(|b| b.recovered_block().hash()))
            .collect::<Vec<_>>();
        for hash in blocks_to_remove {
            if let Some((removed, _)) = self.remove_by_hash(hash) {
                debug!(target: "engine::tree", num_hash=?removed.recovered_block().num_hash(), "Removed finalized sidechain block");
            }
        }

        // remove trie updates that are below the finalized block
        self.persisted_trie_updates.retain(|_, (block_num, _)| *block_num > finalized_num);

        // The only block that should remain at the `finalized` number now, is the finalized
        // block, if it exists.
        //
        // For all other blocks, we  first put their children into this vec.
        // Then, we will iterate over them, removing them, adding their children, etc etc,
        // until the vec is empty.
        let mut blocks_to_remove = self.blocks_by_number.remove(&finalized_num).unwrap_or_default();

        // re-insert the finalized hash if we removed it
        if let Some(position) =
            blocks_to_remove.iter().position(|b| b.recovered_block().hash() == finalized_hash)
        {
            let finalized_block = blocks_to_remove.swap_remove(position);
            self.blocks_by_number.insert(finalized_num, vec![finalized_block]);
        }

        let mut blocks_to_remove = blocks_to_remove
            .into_iter()
            .map(|e| e.recovered_block().hash())
            .collect::<VecDeque<_>>();
        while let Some(block) = blocks_to_remove.pop_front() {
            if let Some((removed, children)) = self.remove_by_hash(block) {
                debug!(target: "engine::tree", num_hash=?removed.recovered_block().num_hash(), "Removed finalized sidechain child block");
                blocks_to_remove.extend(children);
            }
        }
    }

    /// Remove all blocks up to __and including__ the given block number.
    ///
    /// If a finalized hash is provided, the only non-canonical blocks which will be removed are
    /// those which have a fork point at or below the finalized hash.
    ///
    /// Canonical blocks below the upper bound will still be removed.
    ///
    /// NOTE: if the finalized block is greater than the upper bound, the only blocks that will be
    /// removed are canonical blocks and sidechains that fork below the `upper_bound`. This is the
    /// same behavior as if the `finalized_num` were `Some(upper_bound)`.
    pub(crate) fn remove_until(
        &mut self,
        upper_bound: BlockNumHash,
        last_persisted_hash: B256,
        finalized_num_hash: Option<BlockNumHash>,
    ) {
        debug!(target: "engine::tree", ?upper_bound, ?finalized_num_hash, "Removing blocks from the tree");

        // If the finalized num is ahead of the upper bound, and exists, we need to instead ensure
        // that the only blocks removed, are canonical blocks less than the upper bound
        let finalized_num_hash = finalized_num_hash.map(|mut finalized| {
            if upper_bound.number < finalized.number {
                finalized = upper_bound;
                debug!(target: "engine::tree", ?finalized, "Adjusted upper bound");
            }
            finalized
        });

        // We want to do two things:
        // * remove canonical blocks that are persisted
        // * remove forks whose root are below the finalized block
        // We can do this in 2 steps:
        // * remove all canonical blocks below the upper bound
        // * fetch the number of the finalized hash, removing any sidechains that are __below__ the
        // finalized block
        self.remove_canonical_until(upper_bound.number, last_persisted_hash);

        // Now, we have removed canonical blocks (assuming the upper bound is above the finalized
        // block) and only have sidechains below the finalized block.
        if let Some(finalized_num_hash) = finalized_num_hash {
            self.prune_finalized_sidechains(finalized_num_hash);
        }
    }

    /// Determines if the second block is a direct descendant of the first block.
    ///
    /// If the two blocks are the same, this returns `false`.
    fn is_descendant(&self, first: BlockNumHash, second: &N::BlockHeader) -> bool {
        // If the second block's parent is the first block's hash, then it is a direct descendant
        // and we can return early.
        if second.parent_hash() == first.hash {
            return true
        }

        // If the second block is lower than, or has the same block number, they are not
        // descendants.
        if second.number() <= first.number {
            return false
        }

        // iterate through parents of the second until we reach the number
        let Some(mut current_block) = self.block_by_hash(second.parent_hash()) else {
            // If we can't find its parent in the tree, we can't continue, so return false
            return false
        };

        while current_block.number() > first.number + 1 {
            let Some(block) = self.block_by_hash(current_block.header().parent_hash()) else {
                // If we can't find its parent in the tree, we can't continue, so return false
                return false
            };

            current_block = block;
        }

        // Now the block numbers should be equal, so we compare hashes.
        current_block.parent_hash() == first.hash
    }

    /// Updates the canonical head to the given block.
    fn set_canonical_head(&mut self, new_head: BlockNumHash) {
        self.current_canonical_head = new_head;
    }

    /// Returns the tracked canonical head.
    const fn canonical_head(&self) -> &BlockNumHash {
        &self.current_canonical_head
    }

    /// Returns the block hash of the canonical head.
    const fn canonical_block_hash(&self) -> B256 {
        self.canonical_head().hash
    }

    /// Returns the block number of the canonical head.
    const fn canonical_block_number(&self) -> BlockNumber {
        self.canonical_head().number
    }
}

/// A builder for creating state providers that can be used across threads.
#[derive(Clone, Debug)]
pub struct StateProviderBuilder<N: NodePrimitives, P> {
    /// The provider factory used to create providers.
    provider_factory: P,
    /// The historical block hash to fetch state from.
    historical: B256,
    /// The blocks that form the chain from historical to target and are in memory.
    overlay: Option<Vec<ExecutedBlockWithTrieUpdates<N>>>,
}

impl<N: NodePrimitives, P> StateProviderBuilder<N, P> {
    /// Creates a new state provider from the provider factory, historical block hash and optional
    /// overlaid blocks.
    pub fn new(
        provider_factory: P,
        historical: B256,
        overlay: Option<Vec<ExecutedBlockWithTrieUpdates<N>>>,
    ) -> Self {
        Self { provider_factory, historical, overlay }
    }
}

impl<N: NodePrimitives, P> StateProviderBuilder<N, P>
where
    P: BlockReader + StateProviderFactory + StateReader + StateCommitmentProvider + Clone,
{
    /// Creates a new state provider from this builder.
    pub fn build(&self) -> ProviderResult<StateProviderBox> {
        let mut provider = self.provider_factory.state_by_block_hash(self.historical)?;
        if let Some(overlay) = self.overlay.clone() {
            provider = Box::new(MemoryOverlayStateProvider::new(provider, overlay))
        }
        Ok(provider)
    }
}

/// Tracks the state of the engine api internals.
///
/// This type is not shareable.
#[derive(Debug)]
pub struct EngineApiTreeState<N: NodePrimitives> {
    /// Tracks the state of the blockchain tree.
    tree_state: TreeState<N>,
    /// Tracks the forkchoice state updates received by the CL.
    forkchoice_state_tracker: ForkchoiceStateTracker,
    /// Buffer of detached blocks.
    buffer: BlockBuffer<N::Block>,
    /// Tracks the header of invalid payloads that were rejected by the engine because they're
    /// invalid.
    invalid_headers: InvalidHeaderCache,
}

impl<N: NodePrimitives> EngineApiTreeState<N> {
    fn new(
        block_buffer_limit: u32,
        max_invalid_header_cache_length: u32,
        canonical_block: BlockNumHash,
    ) -> Self {
        Self {
            invalid_headers: InvalidHeaderCache::new(max_invalid_header_cache_length),
            buffer: BlockBuffer::new(block_buffer_limit),
            tree_state: TreeState::new(canonical_block),
            forkchoice_state_tracker: ForkchoiceStateTracker::default(),
        }
    }
}

/// The outcome of a tree operation.
#[derive(Debug)]
pub struct TreeOutcome<T> {
    /// The outcome of the operation.
    pub outcome: T,
    /// An optional event to tell the caller to do something.
    pub event: Option<TreeEvent>,
}

impl<T> TreeOutcome<T> {
    /// Create new tree outcome.
    pub const fn new(outcome: T) -> Self {
        Self { outcome, event: None }
    }

    /// Set event on the outcome.
    pub fn with_event(mut self, event: TreeEvent) -> Self {
        self.event = Some(event);
        self
    }
}

/// Events that are triggered by Tree Chain
#[derive(Debug)]
pub enum TreeEvent {
    /// Tree action is needed.
    TreeAction(TreeAction),
    /// Backfill action is needed.
    BackfillAction(BackfillAction),
    /// Block download is needed.
    Download(DownloadRequest),
}

impl TreeEvent {
    /// Returns true if the event is a backfill action.
    const fn is_backfill_action(&self) -> bool {
        matches!(self, Self::BackfillAction(_))
    }
}

/// The actions that can be performed on the tree.
#[derive(Debug)]
pub enum TreeAction {
    /// Make target canonical.
    MakeCanonical {
        /// The sync target head hash
        sync_target_head: B256,
    },
}

/// The engine API tree handler implementation.
///
/// This type is responsible for processing engine API requests, maintaining the canonical state and
/// emitting events.
pub struct EngineApiTreeHandler<N, P, E, T, V, C>
where
    N: NodePrimitives,
    T: EngineTypes,
{
    provider: P,
    executor_provider: E,
    evm_config: C,
    consensus: Arc<dyn FullConsensus<N, Error = ConsensusError>>,
    payload_validator: V,
    /// Keeps track of internals such as executed and buffered blocks.
    state: EngineApiTreeState<N>,
    /// The half for sending messages to the engine.
    ///
    /// This is kept so that we can queue in messages to ourself that we can process later, for
    /// example distributing workload across multiple messages that would otherwise take too long
    /// to process. E.g. we might receive a range of downloaded blocks and we want to process
    /// them one by one so that we can handle incoming engine API in between and don't become
    /// unresponsive. This can happen during live sync transition where we're trying to close the
    /// gap (up to 3 epochs of blocks in the worst case).
    incoming_tx: Sender<FromEngine<EngineApiRequest<T, N>, N::Block>>,
    /// Incoming engine API requests.
    incoming: Receiver<FromEngine<EngineApiRequest<T, N>, N::Block>>,
    /// Outgoing events that are emitted to the handler.
    outgoing: UnboundedSender<EngineApiEvent<N>>,
    /// Channels to the persistence layer.
    persistence: PersistenceHandle<N>,
    /// Tracks the state changes of the persistence task.
    persistence_state: PersistenceState,
    /// Flag indicating the state of the node's backfill synchronization process.
    backfill_sync_state: BackfillSyncState,
    /// Keeps track of the state of the canonical chain that isn't persisted yet.
    /// This is intended to be accessed from external sources, such as rpc.
    canonical_in_memory_state: CanonicalInMemoryState<N>,
    /// Handle to the payload builder that will receive payload attributes for valid forkchoice
    /// updates
    payload_builder: PayloadBuilderHandle<T>,
    /// Configuration settings.
    config: TreeConfig,
    /// Metrics for the engine api.
    metrics: EngineApiMetrics,
    /// An invalid block hook.
    invalid_block_hook: Box<dyn InvalidBlockHook<N>>,
    /// The engine API variant of this handler
    engine_kind: EngineApiKind,
    /// The type responsible for processing new payloads
    payload_processor: PayloadProcessor<N, C>,
}

impl<N, P: Debug, E: Debug, T: EngineTypes + Debug, V: Debug, C: Debug> std::fmt::Debug
    for EngineApiTreeHandler<N, P, E, T, V, C>
where
    N: NodePrimitives,
{
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("EngineApiTreeHandler")
            .field("provider", &self.provider)
            .field("evm_config", &self.evm_config)
            .field("executor_provider", &self.executor_provider)
            .field("consensus", &self.consensus)
            .field("payload_validator", &self.payload_validator)
            .field("state", &self.state)
            .field("incoming_tx", &self.incoming_tx)
            .field("persistence", &self.persistence)
            .field("persistence_state", &self.persistence_state)
            .field("backfill_sync_state", &self.backfill_sync_state)
            .field("canonical_in_memory_state", &self.canonical_in_memory_state)
            .field("payload_builder", &self.payload_builder)
            .field("config", &self.config)
            .field("metrics", &self.metrics)
            .field("invalid_block_hook", &format!("{:p}", self.invalid_block_hook))
            .field("engine_kind", &self.engine_kind)
            .finish()
    }
}

impl<N, P, E, T, V, C> EngineApiTreeHandler<N, P, E, T, V, C>
where
    N: NodePrimitives,
    P: DatabaseProviderFactory
        + BlockReader<Block = N::Block, Header = N::BlockHeader>
        + StateProviderFactory
        + StateReader<Receipt = N::Receipt>
        + StateCommitmentProvider
        + HashedPostStateProvider
        + Clone
        + 'static,
    <P as DatabaseProviderFactory>::Provider:
        BlockReader<Block = N::Block, Header = N::BlockHeader>,
    E: BlockExecutorProvider<Primitives = N>,
    C: ConfigureEvm<Primitives = N> + 'static,
    T: EngineTypes,
    V: EngineValidator<T, Block = N::Block>,
{
    /// Creates a new [`EngineApiTreeHandler`].
    #[expect(clippy::too_many_arguments)]
    pub fn new(
        provider: P,
        executor_provider: E,
        consensus: Arc<dyn FullConsensus<N, Error = ConsensusError>>,
        payload_validator: V,
        outgoing: UnboundedSender<EngineApiEvent<N>>,
        state: EngineApiTreeState<N>,
        canonical_in_memory_state: CanonicalInMemoryState<N>,
        persistence: PersistenceHandle<N>,
        persistence_state: PersistenceState,
        payload_builder: PayloadBuilderHandle<T>,
        config: TreeConfig,
        engine_kind: EngineApiKind,
        evm_config: C,
    ) -> Self {
        let (incoming_tx, incoming) = std::sync::mpsc::channel();

        let payload_processor =
            PayloadProcessor::new(WorkloadExecutor::default(), evm_config.clone(), &config);

        Self {
            provider,
            executor_provider,
            evm_config,
            consensus,
            payload_validator,
            incoming,
            outgoing,
            persistence,
            persistence_state,
            backfill_sync_state: BackfillSyncState::Idle,
            state,
            canonical_in_memory_state,
            payload_builder,
            config,
            metrics: Default::default(),
            incoming_tx,
            invalid_block_hook: Box::new(NoopInvalidBlockHook),
            engine_kind,
            payload_processor,
        }
    }

    /// Sets the invalid block hook.
    fn set_invalid_block_hook(&mut self, invalid_block_hook: Box<dyn InvalidBlockHook<N>>) {
        self.invalid_block_hook = invalid_block_hook;
    }

    /// Creates a new [`EngineApiTreeHandler`] instance and spawns it in its
    /// own thread.
    ///
    /// Returns the sender through which incoming requests can be sent to the task and the receiver
    /// end of a [`EngineApiEvent`] unbounded channel to receive events from the engine.
    #[expect(clippy::complexity)]
    pub fn spawn_new(
        provider: P,
        executor_provider: E,
        consensus: Arc<dyn FullConsensus<N, Error = ConsensusError>>,
        payload_validator: V,
        persistence: PersistenceHandle<N>,
        payload_builder: PayloadBuilderHandle<T>,
        canonical_in_memory_state: CanonicalInMemoryState<N>,
        config: TreeConfig,
        invalid_block_hook: Box<dyn InvalidBlockHook<N>>,
        kind: EngineApiKind,
        evm_config: C,
    ) -> (Sender<FromEngine<EngineApiRequest<T, N>, N::Block>>, UnboundedReceiver<EngineApiEvent<N>>)
    {
        let best_block_number = provider.best_block_number().unwrap_or(0);
        let header = provider.sealed_header(best_block_number).ok().flatten().unwrap_or_default();

        let persistence_state = PersistenceState {
            last_persisted_block: BlockNumHash::new(best_block_number, header.hash()),
            rx: None,
            remove_above_state: VecDeque::new(),
        };

        let (tx, outgoing) = unbounded_channel();
        let state = EngineApiTreeState::new(
            config.block_buffer_limit(),
            config.max_invalid_header_cache_length(),
            header.num_hash(),
        );

        let mut task = Self::new(
            provider,
            executor_provider,
            consensus,
            payload_validator,
            tx,
            state,
            canonical_in_memory_state,
            persistence,
            persistence_state,
            payload_builder,
            config,
            kind,
            evm_config,
        );
        task.set_invalid_block_hook(invalid_block_hook);
        let incoming = task.incoming_tx.clone();
        std::thread::Builder::new().name("Tree Task".to_string()).spawn(|| task.run()).unwrap();
        (incoming, outgoing)
    }

    /// Returns a new [`Sender`] to send messages to this type.
    pub fn sender(&self) -> Sender<FromEngine<EngineApiRequest<T, N>, N::Block>> {
        self.incoming_tx.clone()
    }

    /// Run the engine API handler.
    ///
    /// This will block the current thread and process incoming messages.
    pub fn run(mut self) {
        loop {
            match self.try_recv_engine_message() {
                Ok(Some(msg)) => {
                    debug!(target: "engine::tree", %msg, "received new engine message");
                    if let Err(fatal) = self.on_engine_message(msg) {
                        error!(target: "engine::tree", %fatal, "insert block fatal error");
                        return
                    }
                }
                Ok(None) => {
                    debug!(target: "engine::tree", "received no engine message for some time, while waiting for persistence task to complete");
                }
                Err(_err) => {
                    error!(target: "engine::tree", "Engine channel disconnected");
                    return
                }
            }

            if let Err(err) = self.advance_persistence() {
                error!(target: "engine::tree", %err, "Advancing persistence failed");
                return
            }
        }
    }

    /// Invoked when previously requested blocks were downloaded.
    ///
    /// If the block count exceeds the configured batch size we're allowed to execute at once, this
    /// will execute the first batch and send the remaining blocks back through the channel so that
    /// block request processing isn't blocked for a long time.
    fn on_downloaded(
        &mut self,
        mut blocks: Vec<RecoveredBlock<N::Block>>,
    ) -> Result<Option<TreeEvent>, InsertBlockFatalError> {
        if blocks.is_empty() {
            // nothing to execute
            return Ok(None)
        }

        trace!(target: "engine::tree", block_count = %blocks.len(), "received downloaded blocks");
        let batch = self.config.max_execute_block_batch_size().min(blocks.len());
        for block in blocks.drain(..batch) {
            if let Some(event) = self.on_downloaded_block(block)? {
                let needs_backfill = event.is_backfill_action();
                self.on_tree_event(event)?;
                if needs_backfill {
                    // can exit early if backfill is needed
                    return Ok(None)
                }
            }
        }

        // if we still have blocks to execute, send them as a followup request
        if !blocks.is_empty() {
            let _ = self.incoming_tx.send(FromEngine::DownloadedBlocks(blocks));
        }

        Ok(None)
    }

    /// When the Consensus layer receives a new block via the consensus gossip protocol,
    /// the transactions in the block are sent to the execution layer in the form of a
    /// [`EngineTypes::ExecutionData`]. The Execution layer executes the transactions and validates
    /// the state in the block header, then passes validation data back to Consensus layer, that
    /// adds the block to the head of its own blockchain and attests to it. The block is then
    /// broadcast over the consensus p2p network in the form of a "Beacon block".
    ///
    /// These responses should adhere to the [Engine API Spec for
    /// `engine_newPayload`](https://github.com/ethereum/execution-apis/blob/main/src/engine/paris.md#specification).
    ///
    /// This returns a [`PayloadStatus`] that represents the outcome of a processed new payload and
    /// returns an error if an internal error occurred.
    #[instrument(level = "trace", skip_all, fields(block_hash = %payload.block_hash(), block_num = %payload.block_number(),), target = "engine::tree")]
    fn on_new_payload(
        &mut self,
        payload: T::ExecutionData,
    ) -> Result<TreeOutcome<PayloadStatus>, InsertBlockFatalError> {
        trace!(target: "engine::tree", "invoked new payload");
        self.metrics.engine.new_payload_messages.increment(1);

        // Ensures that the given payload does not violate any consensus rules that concern the
        // block's layout, like:
        //    - missing or invalid base fee
        //    - invalid extra data
        //    - invalid transactions
        //    - incorrect hash
        //    - the versioned hashes passed with the payload do not exactly match transaction
        //      versioned hashes
        //    - the block does not contain blob transactions if it is pre-cancun
        //
        // This validates the following engine API rule:
        //
        // 3. Given the expected array of blob versioned hashes client software **MUST** run its
        //    validation by taking the following steps:
        //
        //   1. Obtain the actual array by concatenating blob versioned hashes lists
        //      (`tx.blob_versioned_hashes`) of each [blob
        //      transaction](https://eips.ethereum.org/EIPS/eip-4844#new-transaction-type) included
        //      in the payload, respecting the order of inclusion. If the payload has no blob
        //      transactions the expected array **MUST** be `[]`.
        //
        //   2. Return `{status: INVALID, latestValidHash: null, validationError: errorMessage |
        //      null}` if the expected and the actual arrays don't match.
        //
        // This validation **MUST** be instantly run in all cases even during active sync process.
        let parent_hash = payload.parent_hash();
        let block = match self.payload_validator.ensure_well_formed_payload(payload) {
            Ok(block) => block,
            Err(error) => {
                error!(target: "engine::tree", %error, "Invalid payload");
                // we need to convert the error to a payload status (response to the CL)

                let latest_valid_hash =
                    if error.is_block_hash_mismatch() || error.is_invalid_versioned_hashes() {
                        // Engine-API rules:
                        // > `latestValidHash: null` if the blockHash validation has failed (<https://github.com/ethereum/execution-apis/blob/fe8e13c288c592ec154ce25c534e26cb7ce0530d/src/engine/shanghai.md?plain=1#L113>)
                        // > `latestValidHash: null` if the expected and the actual arrays don't match (<https://github.com/ethereum/execution-apis/blob/fe8e13c288c592ec154ce25c534e26cb7ce0530d/src/engine/cancun.md?plain=1#L103>)
                        None
                    } else {
                        self.latest_valid_hash_for_invalid_payload(parent_hash)?
                    };

                let status = PayloadStatusEnum::from(error);
                return Ok(TreeOutcome::new(PayloadStatus::new(status, latest_valid_hash)))
            }
        };

        let block_hash = block.hash();
        let mut lowest_buffered_ancestor = self.lowest_buffered_ancestor_or(block_hash);
        if lowest_buffered_ancestor == block_hash {
            lowest_buffered_ancestor = block.parent_hash();
        }

        // now check the block itself
        if let Some(status) =
            self.check_invalid_ancestor_with_head(lowest_buffered_ancestor, &block)?
        {
            return Ok(TreeOutcome::new(status))
        }

        let status = if self.backfill_sync_state.is_idle() {
            let mut latest_valid_hash = None;
            let num_hash = block.num_hash();
            match self.insert_block(block) {
                Ok(status) => {
                    let status = match status {
                        InsertPayloadOk::Inserted(BlockStatus::Valid) => {
                            latest_valid_hash = Some(block_hash);
                            self.try_connect_buffered_blocks(num_hash)?;
                            PayloadStatusEnum::Valid
                        }
                        InsertPayloadOk::AlreadySeen(BlockStatus::Valid) => {
                            latest_valid_hash = Some(block_hash);
                            PayloadStatusEnum::Valid
                        }
                        InsertPayloadOk::Inserted(BlockStatus::Disconnected { .. }) |
                        InsertPayloadOk::AlreadySeen(BlockStatus::Disconnected { .. }) => {
                            // not known to be invalid, but we don't know anything else
                            PayloadStatusEnum::Syncing
                        }
                    };

                    PayloadStatus::new(status, latest_valid_hash)
                }
                Err(error) => self.on_insert_block_error(error)?,
            }
        } else if let Err(error) = self.buffer_block(block) {
            self.on_insert_block_error(error)?
        } else {
            PayloadStatus::from_status(PayloadStatusEnum::Syncing)
        };

        let mut outcome = TreeOutcome::new(status);
        // if the block is valid and it is the current sync target head, make it canonical
        if outcome.outcome.is_valid() && self.is_sync_target_head(block_hash) {
            // but only if it isn't already the canonical head
            if self.state.tree_state.canonical_block_hash() != block_hash {
                outcome = outcome.with_event(TreeEvent::TreeAction(TreeAction::MakeCanonical {
                    sync_target_head: block_hash,
                }));
            }
        }

        Ok(outcome)
    }

    /// Returns the new chain for the given head.
    ///
    /// This also handles reorgs.
    ///
    /// Note: This does not update the tracked state and instead returns the new chain based on the
    /// given head.
    fn on_new_head(&self, new_head: B256) -> ProviderResult<Option<NewCanonicalChain<N>>> {
        // get the executed new head block
        let Some(new_head_block) = self.state.tree_state.blocks_by_hash.get(&new_head) else {
            return Ok(None)
        };

        let new_head_number = new_head_block.recovered_block().number();
        let mut current_canonical_number = self.state.tree_state.current_canonical_head.number;

        let mut new_chain = vec![new_head_block.clone()];
        let mut current_hash = new_head_block.recovered_block().parent_hash();
        let mut current_number = new_head_number - 1;

        // Walk back the new chain until we reach a block we know about
        //
        // This is only done for in-memory blocks, because we should not have persisted any blocks
        // that are _above_ the current canonical head.
        while current_number > current_canonical_number {
            if let Some(block) = self.state.tree_state.executed_block_by_hash(current_hash).cloned()
            {
                current_hash = block.recovered_block().parent_hash();
                current_number -= 1;
                new_chain.push(block);
            } else {
                warn!(target: "engine::tree", current_hash=?current_hash, "Sidechain block not found in TreeState");
                // This should never happen as we're walking back a chain that should connect to
                // the canonical chain
                return Ok(None);
            }
        }

        // If we have reached the current canonical head by walking back from the target, then we
        // know this represents an extension of the canonical chain.
        if current_hash == self.state.tree_state.current_canonical_head.hash {
            new_chain.reverse();

            // Simple extension of the current chain
            return Ok(Some(NewCanonicalChain::Commit { new: new_chain }));
        }

        // We have a reorg. Walk back both chains to find the fork point.
        let mut old_chain = Vec::new();
        let mut old_hash = self.state.tree_state.current_canonical_head.hash;

        // If the canonical chain is ahead of the new chain,
        // gather all blocks until new head number.
        while current_canonical_number > current_number {
            if let Some(block) = self.canonical_block_by_hash(old_hash)? {
                old_chain.push(block.clone());
                old_hash = block.recovered_block().parent_hash();
                current_canonical_number -= 1;
            } else {
                // This shouldn't happen as we're walking back the canonical chain
                warn!(target: "engine::tree", current_hash=?old_hash, "Canonical block not found in TreeState");
                return Ok(None);
            }
        }

        // Both new and old chain pointers are now at the same height.
        debug_assert_eq!(current_number, current_canonical_number);

        // Walk both chains from specified hashes at same height until
        // a common ancestor (fork block) is reached.
        while old_hash != current_hash {
            if let Some(block) = self.canonical_block_by_hash(old_hash)? {
                old_hash = block.recovered_block().parent_hash();
                old_chain.push(block);
            } else {
                // This shouldn't happen as we're walking back the canonical chain
                warn!(target: "engine::tree", current_hash=?old_hash, "Canonical block not found in TreeState");
                return Ok(None);
            }

            if let Some(block) = self.state.tree_state.executed_block_by_hash(current_hash).cloned()
            {
                current_hash = block.recovered_block().parent_hash();
                new_chain.push(block);
            } else {
                // This shouldn't happen as we've already walked this path
                warn!(target: "engine::tree", invalid_hash=?current_hash, "New chain block not found in TreeState");
                return Ok(None);
            }
        }
        new_chain.reverse();
        old_chain.reverse();

        Ok(Some(NewCanonicalChain::Reorg { new: new_chain, old: old_chain }))
    }

    /// Determines if the given block is part of a fork by checking that these
    /// conditions are true:
    /// * walking back from the target hash to verify that the target hash is not part of an
    ///   extension of the canonical chain.
    /// * walking back from the current head to verify that the target hash is not already part of
    ///   the canonical chain.
    fn is_fork(&self, target_hash: B256) -> ProviderResult<bool> {
        // verify that the given hash is not part of an extension of the canon chain.
        let canonical_head = self.state.tree_state.canonical_head();
        let mut current_hash = target_hash;
        while let Some(current_block) = self.sealed_header_by_hash(current_hash)? {
            if current_block.hash() == canonical_head.hash {
                return Ok(false)
            }
            // We already passed the canonical head
            if current_block.number() <= canonical_head.number {
                break
            }
            current_hash = current_block.parent_hash();
        }

        // verify that the given hash is not already part of canonical chain stored in memory
        if self.canonical_in_memory_state.header_by_hash(target_hash).is_some() {
            return Ok(false)
        }

        // verify that the given hash is not already part of persisted canonical chain
        if self.provider.block_number(target_hash)?.is_some() {
            return Ok(false)
        }

        Ok(true)
    }

    /// Returns the persisting kind for the input block.
    fn persisting_kind_for(&self, block: &N::BlockHeader) -> PersistingKind {
        // Check that we're currently persisting.
        let Some(action) = self.persistence_state.current_action() else {
            return PersistingKind::NotPersisting
        };
        // Check that the persistince action is saving blocks, not removing them.
        let CurrentPersistenceAction::SavingBlocks { highest } = action else {
            return PersistingKind::PersistingNotDescendant
        };

        // The block being validated can only be a descendant if its number is higher than
        // the highest block persisting. Otherwise, it's likely a fork of a lower block.
        if block.number() > highest.number && self.state.tree_state.is_descendant(*highest, block) {
            return PersistingKind::PersistingDescendant
        }

        // In all other cases, the block is not a descendant.
        PersistingKind::PersistingNotDescendant
    }

    /// Invoked when we receive a new forkchoice update message. Calls into the blockchain tree
    /// to resolve chain forks and ensure that the Execution Layer is working with the latest valid
    /// chain.
    ///
    /// These responses should adhere to the [Engine API Spec for
    /// `engine_forkchoiceUpdated`](https://github.com/ethereum/execution-apis/blob/main/src/engine/paris.md#specification-1).
    ///
    /// Returns an error if an internal error occurred like a database error.
    #[instrument(level = "trace", skip_all, fields(head = % state.head_block_hash, safe = % state.safe_block_hash,finalized = % state.finalized_block_hash), target = "engine::tree")]
    fn on_forkchoice_updated(
        &mut self,
        state: ForkchoiceState,
        attrs: Option<T::PayloadAttributes>,
        version: EngineApiMessageVersion,
    ) -> ProviderResult<TreeOutcome<OnForkChoiceUpdated>> {
        trace!(target: "engine::tree", ?attrs, "invoked forkchoice update");
        self.metrics.engine.forkchoice_updated_messages.increment(1);
        self.canonical_in_memory_state.on_forkchoice_update_received();

        if let Some(on_updated) = self.pre_validate_forkchoice_update(state)? {
            return Ok(TreeOutcome::new(on_updated))
        }

        let valid_outcome = |head| {
            TreeOutcome::new(OnForkChoiceUpdated::valid(PayloadStatus::new(
                PayloadStatusEnum::Valid,
                Some(head),
            )))
        };

        // Process the forkchoice update by trying to make the head block canonical
        //
        // We can only process this forkchoice update if:
        // - we have the `head` block
        // - the head block is part of a chain that is connected to the canonical chain. This
        //   includes reorgs.
        //
        // Performing a FCU involves:
        // - marking the FCU's head block as canonical
        // - updating in memory state to reflect the new canonical chain
        // - updating canonical state trackers
        // - emitting a canonicalization event for the new chain (including reorg)
        // - if we have payload attributes, delegate them to the payload service

        // 1. ensure we have a new head block
        if self.state.tree_state.canonical_block_hash() == state.head_block_hash {
            trace!(target: "engine::tree", "fcu head hash is already canonical");

            // update the safe and finalized blocks and ensure their values are valid
            if let Err(outcome) = self.ensure_consistent_forkchoice_state(state) {
                // safe or finalized hashes are invalid
                return Ok(TreeOutcome::new(outcome))
            }

            // we still need to process payload attributes if the head is already canonical
            if let Some(attr) = attrs {
                let tip = self
                    .block_by_hash(self.state.tree_state.canonical_block_hash())?
                    .ok_or_else(|| {
                        // If we can't find the canonical block, then something is wrong and we need
                        // to return an error
                        ProviderError::HeaderNotFound(state.head_block_hash.into())
                    })?;
                let updated = self.process_payload_attributes(attr, tip.header(), state, version);
                return Ok(TreeOutcome::new(updated))
            }

            // the head block is already canonical
            return Ok(valid_outcome(state.head_block_hash))
        }

        // 2. ensure we can apply a new chain update for the head block
        if let Some(chain_update) = self.on_new_head(state.head_block_hash)? {
            let tip = chain_update.tip().clone_sealed_header();
            self.on_canonical_chain_update(chain_update);

            // update the safe and finalized blocks and ensure their values are valid
            if let Err(outcome) = self.ensure_consistent_forkchoice_state(state) {
                // safe or finalized hashes are invalid
                return Ok(TreeOutcome::new(outcome))
            }

            if let Some(attr) = attrs {
                let updated = self.process_payload_attributes(attr, &tip, state, version);
                return Ok(TreeOutcome::new(updated))
            }

            return Ok(valid_outcome(state.head_block_hash))
        }

        // 3. check if the head is already part of the canonical chain
        if let Ok(Some(canonical_header)) = self.find_canonical_header(state.head_block_hash) {
            debug!(target: "engine::tree", head = canonical_header.number(), "fcu head block is already canonical");

            // For OpStack the proposers are allowed to reorg their own chain at will, so we need to
            // always trigger a new payload job if requested.
            if self.engine_kind.is_opstack() {
                if let Some(attr) = attrs {
                    debug!(target: "engine::tree", head = canonical_header.number(), "handling payload attributes for canonical head");
                    let updated =
                        self.process_payload_attributes(attr, &canonical_header, state, version);
                    return Ok(TreeOutcome::new(updated))
                }
            }

            // 2. Client software MAY skip an update of the forkchoice state and MUST NOT begin a
            //    payload build process if `forkchoiceState.headBlockHash` references a `VALID`
            //    ancestor of the head of canonical chain, i.e. the ancestor passed payload
            //    validation process and deemed `VALID`. In the case of such an event, client
            //    software MUST return `{payloadStatus: {status: VALID, latestValidHash:
            //    forkchoiceState.headBlockHash, validationError: null}, payloadId: null}`

            // the head block is already canonical, so we're not triggering a payload job and can
            // return right away
            return Ok(valid_outcome(state.head_block_hash))
        }

        // 4. we don't have the block to perform the update
        // we assume the FCU is valid and at least the head is missing,
        // so we need to start syncing to it
        //
        // find the appropriate target to sync to, if we don't have the safe block hash then we
        // start syncing to the safe block via backfill first
        let target = if self.state.forkchoice_state_tracker.is_empty() &&
            // check that safe block is valid and missing
            !state.safe_block_hash.is_zero() &&
            self.find_canonical_header(state.safe_block_hash).ok().flatten().is_none()
        {
            debug!(target: "engine::tree", "missing safe block on initial FCU, downloading safe block");
            state.safe_block_hash
        } else {
            state.head_block_hash
        };

        let target = self.lowest_buffered_ancestor_or(target);
        trace!(target: "engine::tree", %target, "downloading missing block");

        Ok(TreeOutcome::new(OnForkChoiceUpdated::valid(PayloadStatus::from_status(
            PayloadStatusEnum::Syncing,
        )))
        .with_event(TreeEvent::Download(DownloadRequest::single_block(target))))
    }

    /// Attempts to receive the next engine request.
    ///
    /// If there's currently no persistence action in progress, this will block until a new request
    /// is received. If there's a persistence action in progress, this will try to receive the
    /// next request with a timeout to not block indefinitely and return `Ok(None)` if no request is
    /// received in time.
    ///
    /// Returns an error if the engine channel is disconnected.
    #[expect(clippy::type_complexity)]
    fn try_recv_engine_message(
        &self,
    ) -> Result<Option<FromEngine<EngineApiRequest<T, N>, N::Block>>, RecvError> {
        if self.persistence_state.in_progress() {
            // try to receive the next request with a timeout to not block indefinitely
            match self.incoming.recv_timeout(std::time::Duration::from_millis(500)) {
                Ok(msg) => Ok(Some(msg)),
                Err(err) => match err {
                    RecvTimeoutError::Timeout => Ok(None),
                    RecvTimeoutError::Disconnected => Err(RecvError),
                },
            }
        } else {
            self.incoming.recv().map(Some)
        }
    }

    /// Helper method to remove blocks and set the persistence state. This ensures we keep track of
    /// the current persistence action while we're removing blocks.
    fn remove_blocks(&mut self, new_tip_num: u64) {
        debug!(target: "engine::tree", ?new_tip_num, remove_state=?self.persistence_state.remove_above_state, last_persisted_block_number=?self.persistence_state.last_persisted_block.number, "Removing blocks using persistence task");
        if new_tip_num < self.persistence_state.last_persisted_block.number {
            debug!(target: "engine::tree", ?new_tip_num, "Starting remove blocks job");
            let (tx, rx) = oneshot::channel();
            let _ = self.persistence.remove_blocks_above(new_tip_num, tx);
            self.persistence_state.start_remove(new_tip_num, rx);
        }
    }

    /// Helper method to save blocks and set the persistence state. This ensures we keep track of
    /// the current persistence action while we're saving blocks.
    fn persist_blocks(&mut self, blocks_to_persist: Vec<ExecutedBlockWithTrieUpdates<N>>) {
        if blocks_to_persist.is_empty() {
            debug!(target: "engine::tree", "Returned empty set of blocks to persist");
            return
        }

        // NOTE: checked non-empty above
        let highest_num_hash = blocks_to_persist
            .iter()
            .max_by_key(|block| block.recovered_block().number())
            .map(|b| b.recovered_block().num_hash())
            .expect("Checked non-empty persisting blocks");

        debug!(target: "engine::tree", blocks = ?blocks_to_persist.iter().map(|block| block.recovered_block().num_hash()).collect::<Vec<_>>(), "Persisting blocks");
        let (tx, rx) = oneshot::channel();
        let _ = self.persistence.save_blocks(blocks_to_persist, tx);

        self.persistence_state.start_save(highest_num_hash, rx);
    }

    /// Attempts to advance the persistence state.
    ///
    /// If we're currently awaiting a response this will try to receive the response (non-blocking)
    /// or send a new persistence action if necessary.
    fn advance_persistence(&mut self) -> Result<(), AdvancePersistenceError> {
        if self.persistence_state.in_progress() {
            let (mut rx, start_time, current_action) = self
                .persistence_state
                .rx
                .take()
                .expect("if a persistence task is in progress Receiver must be Some");
            // Check if persistence has complete
            match rx.try_recv() {
                Ok(last_persisted_hash_num) => {
                    self.metrics.engine.persistence_duration.record(start_time.elapsed());
                    let Some(BlockNumHash {
                        hash: last_persisted_block_hash,
                        number: last_persisted_block_number,
                    }) = last_persisted_hash_num
                    else {
                        // if this happened, then we persisted no blocks because we sent an
                        // empty vec of blocks
                        warn!(target: "engine::tree", "Persistence task completed but did not persist any blocks");
                        return Ok(())
                    };

                    debug!(target: "engine::tree", ?last_persisted_block_hash, ?last_persisted_block_number, "Finished persisting, calling finish");
                    self.persistence_state
                        .finish(last_persisted_block_hash, last_persisted_block_number);
                    self.on_new_persisted_block()?;
                }
                Err(TryRecvError::Closed) => return Err(TryRecvError::Closed.into()),
                Err(TryRecvError::Empty) => {
                    self.persistence_state.rx = Some((rx, start_time, current_action))
                }
            }
        }

        if !self.persistence_state.in_progress() {
            if let Some(new_tip_num) = self.persistence_state.remove_above_state.pop_front() {
                self.remove_blocks(new_tip_num)
            } else if self.should_persist() {
                let blocks_to_persist = self.get_canonical_blocks_to_persist();
                self.persist_blocks(blocks_to_persist);
            }
        }

        Ok(())
    }

    /// Handles a message from the engine.
    fn on_engine_message(
        &mut self,
        msg: FromEngine<EngineApiRequest<T, N>, N::Block>,
    ) -> Result<(), InsertBlockFatalError> {
        match msg {
            FromEngine::Event(event) => match event {
                FromOrchestrator::BackfillSyncStarted => {
                    debug!(target: "engine::tree", "received backfill sync started event");
                    self.backfill_sync_state = BackfillSyncState::Active;
                }
                FromOrchestrator::BackfillSyncFinished(ctrl) => {
                    self.on_backfill_sync_finished(ctrl)?;
                }
            },
            FromEngine::Request(request) => {
                match request {
                    EngineApiRequest::InsertExecutedBlock(block) => {
                        let block_num_hash = block.recovered_block().num_hash();
                        debug!(target: "engine::tree", block=?block_num_hash, "inserting already executed block");
                        let now = Instant::now();

                        // if the parent is the canonical head, we can insert the block as the
                        // pending block
                        if self.state.tree_state.canonical_block_hash() ==
                            block.recovered_block().parent_hash()
                        {
                            debug!(target: "engine::tree", pending=?block_num_hash, "updating pending block");
                            self.canonical_in_memory_state.set_pending_block(block.clone());
                        }

                        self.state.tree_state.insert_executed(block.clone());
                        self.metrics.engine.inserted_already_executed_blocks.increment(1);
                        self.emit_event(EngineApiEvent::BeaconConsensus(
                            BeaconConsensusEngineEvent::CanonicalBlockAdded(block, now.elapsed()),
                        ));
                    }
                    EngineApiRequest::Beacon(request) => {
                        match request {
                            BeaconEngineMessage::ForkchoiceUpdated {
                                state,
                                payload_attrs,
                                tx,
                                version,
                            } => {
                                let mut output =
                                    self.on_forkchoice_updated(state, payload_attrs, version);

                                if let Ok(res) = &mut output {
                                    // track last received forkchoice state
                                    self.state
                                        .forkchoice_state_tracker
                                        .set_latest(state, res.outcome.forkchoice_status());

                                    // emit an event about the handled FCU
                                    self.emit_event(BeaconConsensusEngineEvent::ForkchoiceUpdated(
                                        state,
                                        res.outcome.forkchoice_status(),
                                    ));

                                    // handle the event if any
                                    self.on_maybe_tree_event(res.event.take())?;
                                }

                                if let Err(err) =
                                    tx.send(output.map(|o| o.outcome).map_err(Into::into))
                                {
                                    self.metrics
                                        .engine
                                        .failed_forkchoice_updated_response_deliveries
                                        .increment(1);
                                    error!(target: "engine::tree", "Failed to send event: {err:?}");
                                }
                            }
                            BeaconEngineMessage::NewPayload { payload, tx } => {
                                let mut output = self.on_new_payload(payload);

                                let maybe_event =
                                    output.as_mut().ok().and_then(|out| out.event.take());

                                // emit response
                                if let Err(err) =
                                    tx.send(output.map(|o| o.outcome).map_err(|e| {
                                        BeaconOnNewPayloadError::Internal(Box::new(e))
                                    }))
                                {
                                    error!(target: "engine::tree", "Failed to send event: {err:?}");
                                    self.metrics
                                        .engine
                                        .failed_new_payload_response_deliveries
                                        .increment(1);
                                }

                                // handle the event if any
                                self.on_maybe_tree_event(maybe_event)?;
                            }
                            BeaconEngineMessage::TransitionConfigurationExchanged => {
                                // triggering this hook will record that we received a request from
                                // the CL
                                self.canonical_in_memory_state
                                    .on_transition_configuration_exchanged();
                            }
                        }
                    }
                }
            }
            FromEngine::DownloadedBlocks(blocks) => {
                if let Some(event) = self.on_downloaded(blocks)? {
                    self.on_tree_event(event)?;
                }
            }
        }
        Ok(())
    }

    /// Invoked if the backfill sync has finished to target.
    ///
    /// At this point we consider the block synced to the backfill target.
    ///
    /// Checks the tracked finalized block against the block on disk and requests another backfill
    /// run if the distance to the tip exceeds the threshold for another backfill run.
    ///
    /// This will also do the necessary housekeeping of the tree state, this includes:
    ///  - removing all blocks below the backfill height
    ///  - resetting the canonical in-memory state
    ///
    /// In case backfill resulted in an unwind, this will clear the tree state above the unwind
    /// target block.
    fn on_backfill_sync_finished(
        &mut self,
        ctrl: ControlFlow,
    ) -> Result<(), InsertBlockFatalError> {
        debug!(target: "engine::tree", "received backfill sync finished event");
        self.backfill_sync_state = BackfillSyncState::Idle;

        // backfill height is the block number that the backfill finished at
        let mut backfill_height = ctrl.block_number();

        // Pipeline unwound, memorize the invalid block and wait for CL for next sync target.
        if let ControlFlow::Unwind { bad_block, target } = &ctrl {
            warn!(target: "engine::tree", invalid_block=?bad_block, "Bad block detected in unwind");
            // update the `invalid_headers` cache with the new invalid header
            self.state.invalid_headers.insert(**bad_block);

            // if this was an unwind then the target is the new height
            backfill_height = Some(*target);
        }

        // backfill height is the block number that the backfill finished at
        let Some(backfill_height) = backfill_height else { return Ok(()) };

        // state house keeping after backfill sync
        // remove all executed blocks below the backfill height
        //
        // We set the `finalized_num` to `Some(backfill_height)` to ensure we remove all state
        // before that
        let Some(backfill_num_hash) = self
            .provider
            .block_hash(backfill_height)?
            .map(|hash| BlockNumHash { hash, number: backfill_height })
        else {
            debug!(target: "engine::tree", ?ctrl, "Backfill block not found");
            return Ok(())
        };

        if ctrl.is_unwind() {
            // the node reset so we need to clear everything above that height so that backfill
            // height is the new canonical block.
            self.state.tree_state.reset(backfill_num_hash)
        } else {
            self.state.tree_state.remove_until(
                backfill_num_hash,
                self.persistence_state.last_persisted_block.hash,
                Some(backfill_num_hash),
            );
        }

        self.metrics.engine.executed_blocks.set(self.state.tree_state.block_count() as f64);
        self.metrics.tree.canonical_chain_height.set(backfill_height as f64);

        // remove all buffered blocks below the backfill height
        self.state.buffer.remove_old_blocks(backfill_height);
        // we remove all entries because now we're synced to the backfill target and consider this
        // the canonical chain
        self.canonical_in_memory_state.clear_state();

        if let Ok(Some(new_head)) = self.provider.sealed_header(backfill_height) {
            // update the tracked chain height, after backfill sync both the canonical height and
            // persisted height are the same
            self.state.tree_state.set_canonical_head(new_head.num_hash());
            self.persistence_state.finish(new_head.hash(), new_head.number());

            // update the tracked canonical head
            self.canonical_in_memory_state.set_canonical_head(new_head);
        }

        // check if we need to run backfill again by comparing the most recent finalized height to
        // the backfill height
        let Some(sync_target_state) = self.state.forkchoice_state_tracker.sync_target_state()
        else {
            return Ok(())
        };
        if sync_target_state.finalized_block_hash.is_zero() {
            // no finalized block, can't check distance
            return Ok(())
        }
        // get the block number of the finalized block, if we have it
        let newest_finalized = self
            .state
            .buffer
            .block(&sync_target_state.finalized_block_hash)
            .map(|block| block.number());

        // The block number that the backfill finished at - if the progress or newest
        // finalized is None then we can't check the distance anyways.
        //
        // If both are Some, we perform another distance check and return the desired
        // backfill target
        if let Some(backfill_target) =
            ctrl.block_number().zip(newest_finalized).and_then(|(progress, finalized_number)| {
                // Determines whether or not we should run backfill again, in case
                // the new gap is still large enough and requires running backfill again
                self.backfill_sync_target(progress, finalized_number, None)
            })
        {
            // request another backfill run
            self.emit_event(EngineApiEvent::BackfillAction(BackfillAction::Start(
                backfill_target.into(),
            )));
            return Ok(())
        };

        // try to close the gap by executing buffered blocks that are child blocks of the new head
        self.try_connect_buffered_blocks(self.state.tree_state.current_canonical_head)
    }

    /// Attempts to make the given target canonical.
    ///
    /// This will update the tracked canonical in memory state and do the necessary housekeeping.
    fn make_canonical(&mut self, target: B256) -> ProviderResult<()> {
        if let Some(chain_update) = self.on_new_head(target)? {
            self.on_canonical_chain_update(chain_update);
        }

        Ok(())
    }

    /// Convenience function to handle an optional tree event.
    fn on_maybe_tree_event(&mut self, event: Option<TreeEvent>) -> ProviderResult<()> {
        if let Some(event) = event {
            self.on_tree_event(event)?;
        }

        Ok(())
    }

    /// Handles a tree event.
    ///
    /// Returns an error if a [`TreeAction::MakeCanonical`] results in a fatal error.
    fn on_tree_event(&mut self, event: TreeEvent) -> ProviderResult<()> {
        match event {
            TreeEvent::TreeAction(action) => match action {
                TreeAction::MakeCanonical { sync_target_head } => {
                    self.make_canonical(sync_target_head)?;
                }
            },
            TreeEvent::BackfillAction(action) => {
                self.emit_event(EngineApiEvent::BackfillAction(action));
            }
            TreeEvent::Download(action) => {
                self.emit_event(EngineApiEvent::Download(action));
            }
        }

        Ok(())
    }

    /// Emits an outgoing event to the engine.
    fn emit_event(&mut self, event: impl Into<EngineApiEvent<N>>) {
        let event = event.into();

        if event.is_backfill_action() {
            debug_assert_eq!(
                self.backfill_sync_state,
                BackfillSyncState::Idle,
                "backfill action should only be emitted when backfill is idle"
            );

            if self.persistence_state.in_progress() {
                // backfill sync and persisting data are mutually exclusive, so we can't start
                // backfill while we're still persisting
                debug!(target: "engine::tree", "skipping backfill file while persistence task is active");
                return
            }

            self.backfill_sync_state = BackfillSyncState::Pending;
            self.metrics.engine.pipeline_runs.increment(1);
            debug!(target: "engine::tree", "emitting backfill action event");
        }

        let _ = self.outgoing.send(event).inspect_err(
            |err| error!(target: "engine::tree", "Failed to send internal event: {err:?}"),
        );
    }

    /// Returns true if the canonical chain length minus the last persisted
    /// block is greater than or equal to the persistence threshold and
    /// backfill is not running.
    pub const fn should_persist(&self) -> bool {
        if !self.backfill_sync_state.is_idle() {
            // can't persist if backfill is running
            return false
        }

        let min_block = self.persistence_state.last_persisted_block.number;
        self.state.tree_state.canonical_block_number().saturating_sub(min_block) >
            self.config.persistence_threshold()
    }

    /// Returns a batch of consecutive canonical blocks to persist in the range
    /// `(last_persisted_number .. canonical_head - threshold]` . The expected
    /// order is oldest -> newest.
    fn get_canonical_blocks_to_persist(&self) -> Vec<ExecutedBlockWithTrieUpdates<N>> {
        let mut blocks_to_persist = Vec::new();
        let mut current_hash = self.state.tree_state.canonical_block_hash();
        let last_persisted_number = self.persistence_state.last_persisted_block.number;

        let canonical_head_number = self.state.tree_state.canonical_block_number();

        let target_number =
            canonical_head_number.saturating_sub(self.config.memory_block_buffer_target());

        debug!(target: "engine::tree", ?last_persisted_number, ?canonical_head_number, ?target_number, ?current_hash, "Returning canonical blocks to persist");
        while let Some(block) = self.state.tree_state.blocks_by_hash.get(&current_hash) {
            if block.recovered_block().number() <= last_persisted_number {
                break;
            }

            if block.recovered_block().number() <= target_number {
                blocks_to_persist.push(block.clone());
            }

            current_hash = block.recovered_block().parent_hash();
        }

        // reverse the order so that the oldest block comes first
        blocks_to_persist.reverse();

        blocks_to_persist
    }

    /// This clears the blocks from the in-memory tree state that have been persisted to the
    /// database.
    ///
    /// This also updates the canonical in-memory state to reflect the newest persisted block
    /// height.
    ///
    /// Assumes that `finish` has been called on the `persistence_state` at least once
    fn on_new_persisted_block(&mut self) -> ProviderResult<()> {
        let finalized = self.state.forkchoice_state_tracker.last_valid_finalized();
        self.remove_before(self.persistence_state.last_persisted_block, finalized)?;
        self.canonical_in_memory_state.remove_persisted_blocks(BlockNumHash {
            number: self.persistence_state.last_persisted_block.number,
            hash: self.persistence_state.last_persisted_block.hash,
        });
        Ok(())
    }

    /// Return an [`ExecutedBlock`] from database or in-memory state by hash.
    ///
    /// NOTE: This cannot fetch [`ExecutedBlock`]s for _finalized_ blocks, instead it can only
    /// fetch [`ExecutedBlock`]s for _canonical_ blocks, or blocks from sidechains that the node
    /// has in memory.
    ///
    /// For finalized blocks, this will return `None`.
    fn canonical_block_by_hash(&self, hash: B256) -> ProviderResult<Option<ExecutedBlock<N>>> {
        trace!(target: "engine::tree", ?hash, "Fetching executed block by hash");
        // check memory first
        if let Some(block) = self.state.tree_state.executed_block_by_hash(hash).cloned() {
            return Ok(Some(block.block))
        }

        let (block, senders) = self
            .provider
            .sealed_block_with_senders(hash.into(), TransactionVariant::WithHash)?
            .ok_or_else(|| ProviderError::HeaderNotFound(hash.into()))?
            .split_sealed();
        let execution_output = self
            .provider
            .get_state(block.header().number())?
            .ok_or_else(|| ProviderError::StateForNumberNotFound(block.header().number()))?;
        let hashed_state = self.provider.hashed_post_state(execution_output.state());

        Ok(Some(ExecutedBlock {
            recovered_block: Arc::new(RecoveredBlock::new_sealed(block, senders)),
            execution_output: Arc::new(execution_output),
            hashed_state: Arc::new(hashed_state),
        }))
    }

    /// Return sealed block from database or in-memory state by hash.
    fn sealed_header_by_hash(
        &self,
        hash: B256,
    ) -> ProviderResult<Option<SealedHeader<N::BlockHeader>>> {
        // check memory first
        let block = self
            .state
            .tree_state
            .block_by_hash(hash)
            .map(|block| block.as_ref().clone_sealed_header());

        if block.is_some() {
            Ok(block)
        } else {
            self.provider.sealed_header_by_hash(hash)
        }
    }

    /// Return block from database or in-memory state by hash.
    fn block_by_hash(&self, hash: B256) -> ProviderResult<Option<N::Block>> {
        // check database first
        let mut block = self.provider.block_by_hash(hash)?;
        if block.is_none() {
            // Note: it's fine to return the unsealed block because the caller already has
            // the hash
            block = self
                .state
                .tree_state
                .block_by_hash(hash)
                // TODO: clone for compatibility. should we return an Arc here?
                .map(|block| block.as_ref().clone().into_block());
        }
        Ok(block)
    }

    /// Return the parent hash of the lowest buffered ancestor for the requested block, if there
    /// are any buffered ancestors. If there are no buffered ancestors, and the block itself does
    /// not exist in the buffer, this returns the hash that is passed in.
    ///
    /// Returns the parent hash of the block itself if the block is buffered and has no other
    /// buffered ancestors.
    fn lowest_buffered_ancestor_or(&self, hash: B256) -> B256 {
        self.state
            .buffer
            .lowest_ancestor(&hash)
            .map(|block| block.parent_hash())
            .unwrap_or_else(|| hash)
    }

    /// If validation fails, the response MUST contain the latest valid hash:
    ///
    ///   - The block hash of the ancestor of the invalid payload satisfying the following two
    ///     conditions:
    ///     - It is fully validated and deemed VALID
    ///     - Any other ancestor of the invalid payload with a higher blockNumber is INVALID
    ///   - 0x0000000000000000000000000000000000000000000000000000000000000000 if the above
    ///     conditions are satisfied by a `PoW` block.
    ///   - null if client software cannot determine the ancestor of the invalid payload satisfying
    ///     the above conditions.
    fn latest_valid_hash_for_invalid_payload(
        &mut self,
        parent_hash: B256,
    ) -> ProviderResult<Option<B256>> {
        // Check if parent exists in side chain or in canonical chain.
        if self.block_by_hash(parent_hash)?.is_some() {
            return Ok(Some(parent_hash))
        }

        // iterate over ancestors in the invalid cache
        // until we encounter the first valid ancestor
        let mut current_hash = parent_hash;
        let mut current_block = self.state.invalid_headers.get(&current_hash);
        while let Some(block_with_parent) = current_block {
            current_hash = block_with_parent.parent;
            current_block = self.state.invalid_headers.get(&current_hash);

            // If current_header is None, then the current_hash does not have an invalid
            // ancestor in the cache, check its presence in blockchain tree
            if current_block.is_none() && self.block_by_hash(current_hash)?.is_some() {
                return Ok(Some(current_hash))
            }
        }
        Ok(None)
    }

    /// Prepares the invalid payload response for the given hash, checking the
    /// database for the parent hash and populating the payload status with the latest valid hash
    /// according to the engine api spec.
    fn prepare_invalid_response(&mut self, mut parent_hash: B256) -> ProviderResult<PayloadStatus> {
        // Edge case: the `latestValid` field is the zero hash if the parent block is the terminal
        // PoW block, which we need to identify by looking at the parent's block difficulty
        if let Some(parent) = self.block_by_hash(parent_hash)? {
            if !parent.header().difficulty().is_zero() {
                parent_hash = B256::ZERO;
            }
        }

        let valid_parent_hash = self.latest_valid_hash_for_invalid_payload(parent_hash)?;
        Ok(PayloadStatus::from_status(PayloadStatusEnum::Invalid {
            validation_error: PayloadValidationError::LinksToRejectedPayload.to_string(),
        })
        .with_latest_valid_hash(valid_parent_hash.unwrap_or_default()))
    }

    /// Returns true if the given hash is the last received sync target block.
    ///
    /// See [`ForkchoiceStateTracker::sync_target_state`]
    fn is_sync_target_head(&self, block_hash: B256) -> bool {
        if let Some(target) = self.state.forkchoice_state_tracker.sync_target_state() {
            return target.head_block_hash == block_hash
        }
        false
    }

    /// Checks if the given `check` hash points to an invalid header, inserting the given `head`
    /// block into the invalid header cache if the `check` hash has a known invalid ancestor.
    ///
    /// Returns a payload status response according to the engine API spec if the block is known to
    /// be invalid.
    fn check_invalid_ancestor_with_head(
        &mut self,
        check: B256,
        head: &SealedBlock<N::Block>,
    ) -> ProviderResult<Option<PayloadStatus>> {
        // check if the check hash was previously marked as invalid
        let Some(header) = self.state.invalid_headers.get(&check) else { return Ok(None) };

        // populate the latest valid hash field
        let status = self.prepare_invalid_response(header.parent)?;

        // insert the head block into the invalid header cache
        self.state.invalid_headers.insert_with_invalid_ancestor(head.hash(), header);
        self.emit_event(BeaconConsensusEngineEvent::InvalidBlock(Box::new(head.clone())));

        Ok(Some(status))
    }

    /// Checks if the given `head` points to an invalid header, which requires a specific response
    /// to a forkchoice update.
    fn check_invalid_ancestor(&mut self, head: B256) -> ProviderResult<Option<PayloadStatus>> {
        // check if the head was previously marked as invalid
        let Some(header) = self.state.invalid_headers.get(&head) else { return Ok(None) };
        // populate the latest valid hash field
        Ok(Some(self.prepare_invalid_response(header.parent)?))
    }

    /// Validate if block is correct and satisfies all the consensus rules that concern the header
    /// and block body itself.
    fn validate_block(&self, block: &RecoveredBlock<N::Block>) -> Result<(), ConsensusError> {
        if let Err(e) =
            self.consensus.validate_header_with_total_difficulty(block.header(), U256::MAX)
        {
            error!(
                target: "engine::tree",
                ?block,
                "Failed to validate total difficulty for block {}: {e}",
                block.hash()
            );
            return Err(e)
        }

        if let Err(e) = self.consensus.validate_header(block.sealed_header()) {
            error!(target: "engine::tree", ?block, "Failed to validate header {}: {e}", block.hash());
            return Err(e)
        }

        if let Err(e) = self.consensus.validate_block_pre_execution(block.sealed_block()) {
            error!(target: "engine::tree", ?block, "Failed to validate block {}: {e}", block.hash());
            return Err(e)
        }

        Ok(())
    }

    /// Attempts to connect any buffered blocks that are connected to the given parent hash.
    #[instrument(level = "trace", skip(self), target = "engine::tree")]
    fn try_connect_buffered_blocks(
        &mut self,
        parent: BlockNumHash,
    ) -> Result<(), InsertBlockFatalError> {
        let blocks = self.state.buffer.remove_block_with_children(&parent.hash);

        if blocks.is_empty() {
            // nothing to append
            return Ok(())
        }

        let now = Instant::now();
        let block_count = blocks.len();
        for child in blocks {
            let child_num_hash = child.num_hash();
            match self.insert_block(child) {
                Ok(res) => {
                    debug!(target: "engine::tree", child =?child_num_hash, ?res, "connected buffered block");
                    if self.is_sync_target_head(child_num_hash.hash) &&
                        matches!(res, InsertPayloadOk::Inserted(BlockStatus::Valid))
                    {
                        self.make_canonical(child_num_hash.hash)?;
                    }
                }
                Err(err) => {
                    debug!(target: "engine::tree", ?err, "failed to connect buffered block to tree");
                    if let Err(fatal) = self.on_insert_block_error(err) {
                        warn!(target: "engine::tree", %fatal, "fatal error occurred while connecting buffered blocks");
                        return Err(fatal)
                    }
                }
            }
        }

        debug!(target: "engine::tree", elapsed = ?now.elapsed(), %block_count, "connected buffered blocks");
        Ok(())
    }

    /// Pre-validates the block and inserts it into the buffer.
    fn buffer_block(
        &mut self,
        block: RecoveredBlock<N::Block>,
    ) -> Result<(), InsertBlockError<N::Block>> {
        if let Err(err) = self.validate_block(&block) {
            return Err(InsertBlockError::consensus_error(err, block.into_sealed_block()))
        }
        self.state.buffer.insert_block(block);
        Ok(())
    }

    /// Returns true if the distance from the local tip to the block is greater than the configured
    /// threshold.
    ///
    /// If the `local_tip` is greater than the `block`, then this will return false.
    #[inline]
    const fn exceeds_backfill_run_threshold(&self, local_tip: u64, block: u64) -> bool {
        block > local_tip && block - local_tip > MIN_BLOCKS_FOR_PIPELINE_RUN
    }

    /// Returns how far the local tip is from the given block. If the local tip is at the same
    /// height or its block number is greater than the given block, this returns None.
    #[inline]
    const fn distance_from_local_tip(&self, local_tip: u64, block: u64) -> Option<u64> {
        if block > local_tip {
            Some(block - local_tip)
        } else {
            None
        }
    }

    /// Returns the target hash to sync to if the distance from the local tip to the block is
    /// greater than the threshold and we're not synced to the finalized block yet (if we've seen
    /// that block already).
    ///
    /// If this is invoked after a new block has been downloaded, the downloaded block could be the
    /// (missing) finalized block.
    fn backfill_sync_target(
        &self,
        canonical_tip_num: u64,
        target_block_number: u64,
        downloaded_block: Option<BlockNumHash>,
    ) -> Option<B256> {
        let sync_target_state = self.state.forkchoice_state_tracker.sync_target_state();

        // check if the distance exceeds the threshold for backfill sync
        let mut exceeds_backfill_threshold =
            self.exceeds_backfill_run_threshold(canonical_tip_num, target_block_number);

        // check if the downloaded block is the tracked finalized block
        if let Some(buffered_finalized) = sync_target_state
            .as_ref()
            .and_then(|state| self.state.buffer.block(&state.finalized_block_hash))
        {
            // if we have buffered the finalized block, we should check how far
            // we're off
            exceeds_backfill_threshold =
                self.exceeds_backfill_run_threshold(canonical_tip_num, buffered_finalized.number());
        }

        // If this is invoked after we downloaded a block we can check if this block is the
        // finalized block
        if let (Some(downloaded_block), Some(ref state)) = (downloaded_block, sync_target_state) {
            if downloaded_block.hash == state.finalized_block_hash {
                // we downloaded the finalized block and can now check how far we're off
                exceeds_backfill_threshold =
                    self.exceeds_backfill_run_threshold(canonical_tip_num, downloaded_block.number);
            }
        }

        // if the number of missing blocks is greater than the max, trigger backfill
        if exceeds_backfill_threshold {
            if let Some(state) = sync_target_state {
                // if we have already canonicalized the finalized block, we should skip backfill
                match self.provider.header_by_hash_or_number(state.finalized_block_hash.into()) {
                    Err(err) => {
                        warn!(target: "engine::tree", %err, "Failed to get finalized block header");
                    }
                    Ok(None) => {
                        // ensure the finalized block is known (not the zero hash)
                        if !state.finalized_block_hash.is_zero() {
                            // we don't have the block yet and the distance exceeds the allowed
                            // threshold
                            return Some(state.finalized_block_hash)
                        }

                        // OPTIMISTIC SYNCING
                        //
                        // It can happen when the node is doing an
                        // optimistic sync, where the CL has no knowledge of the finalized hash,
                        // but is expecting the EL to sync as high
                        // as possible before finalizing.
                        //
                        // This usually doesn't happen on ETH mainnet since CLs use the more
                        // secure checkpoint syncing.
                        //
                        // However, optimism chains will do this. The risk of a reorg is however
                        // low.
                        debug!(target: "engine::tree", hash=?state.head_block_hash, "Setting head hash as an optimistic backfill target.");
                        return Some(state.head_block_hash)
                    }
                    Ok(Some(_)) => {
                        // we're fully synced to the finalized block
                    }
                }
            }
        }

        None
    }

    /// This determines whether or not we should remove blocks from the chain, based on a canonical
    /// chain update.
    ///
    /// If the chain update is a reorg:
    /// * is the new chain behind the last persisted block, or
    /// * if the root of the new chain is at the same height as the last persisted block, is it a
    ///   different block
    ///
    /// If either of these are true, then this returns the height of the first block. Otherwise,
    /// this returns [`None`]. This should be used to check whether or not we should be sending a
    /// remove command to the persistence task.
    fn find_disk_reorg(&self, chain_update: &NewCanonicalChain<N>) -> Option<u64> {
        let NewCanonicalChain::Reorg { new, old: _ } = chain_update else { return None };

        let BlockNumHash { number: new_num, hash: new_hash } =
            new.first().map(|block| block.recovered_block().num_hash())?;

        match new_num.cmp(&self.persistence_state.last_persisted_block.number) {
            Ordering::Greater => {
                // new number is above the last persisted block so the reorg can be performed
                // entirely in memory
                None
            }
            Ordering::Equal => {
                // new number is the same, if the hash is the same then we should not need to remove
                // any blocks
                (self.persistence_state.last_persisted_block.hash != new_hash).then_some(new_num)
            }
            Ordering::Less => {
                // this means we are below the last persisted block and must remove on disk blocks
                Some(new_num)
            }
        }
    }

    /// Invoked when we the canonical chain has been updated.
    ///
    /// This is invoked on a valid forkchoice update, or if we can make the target block canonical.
    fn on_canonical_chain_update(&mut self, chain_update: NewCanonicalChain<N>) {
        trace!(target: "engine::tree", new_blocks = %chain_update.new_block_count(), reorged_blocks =  %chain_update.reorged_block_count(), "applying new chain update");
        let start = Instant::now();

        // schedule a remove_above call if we have an on-disk reorg
        if let Some(height) = self.find_disk_reorg(&chain_update) {
            // calculate the new tip by subtracting one from the lowest part of the chain
            let new_tip_num = height.saturating_sub(1);
            self.persistence_state.schedule_removal(new_tip_num);
        }

        // update the tracked canonical head
        self.state.tree_state.set_canonical_head(chain_update.tip().num_hash());

        let tip = chain_update.tip().clone_sealed_header();
        let notification = chain_update.to_chain_notification();

        // reinsert any missing reorged blocks
        if let NewCanonicalChain::Reorg { new, old } = &chain_update {
            let new_first = new.first().map(|first| first.recovered_block().num_hash());
            let old_first = old.first().map(|first| first.recovered_block().num_hash());
            trace!(target: "engine::tree", ?new_first, ?old_first, "Reorg detected, new and old first blocks");

            self.update_reorg_metrics(old.len());
            self.reinsert_reorged_blocks(new.clone());
            // Try reinserting the reorged canonical chain. This is only possible if we have
            // `persisted_trie_updates` for those blocks.
            let old = old
                .iter()
                .filter_map(|block| {
                    let (_, trie) = self
                        .state
                        .tree_state
                        .persisted_trie_updates
                        .get(&block.recovered_block.hash())
                        .cloned()?;
                    Some(ExecutedBlockWithTrieUpdates { block: block.clone(), trie })
                })
                .collect::<Vec<_>>();
            self.reinsert_reorged_blocks(old);
        }

        // update the tracked in-memory state with the new chain
        self.canonical_in_memory_state.update_chain(chain_update);
        self.canonical_in_memory_state.set_canonical_head(tip.clone());

        // Update metrics based on new tip
        self.metrics.tree.canonical_chain_height.set(tip.number() as f64);

        // sends an event to all active listeners about the new canonical chain
        self.canonical_in_memory_state.notify_canon_state(notification);

        // emit event
        self.emit_event(BeaconConsensusEngineEvent::CanonicalChainCommitted(
            Box::new(tip),
            start.elapsed(),
        ));
    }

    /// This updates metrics based on the given reorg length.
    fn update_reorg_metrics(&self, old_chain_length: usize) {
        self.metrics.tree.reorgs.increment(1);
        self.metrics.tree.latest_reorg_depth.set(old_chain_length as f64);
    }

    /// This reinserts any blocks in the new chain that do not already exist in the tree
    fn reinsert_reorged_blocks(&mut self, new_chain: Vec<ExecutedBlockWithTrieUpdates<N>>) {
        for block in new_chain {
            if self
                .state
                .tree_state
                .executed_block_by_hash(block.recovered_block().hash())
                .is_none()
            {
                trace!(target: "engine::tree", num=?block.recovered_block().number(), hash=?block.recovered_block().hash(), "Reinserting block into tree state");
                self.state.tree_state.insert_executed(block);
            }
        }
    }

    /// Invoke the invalid block hook if this is a new invalid block.
    fn on_invalid_block(
        &mut self,
        parent_header: &SealedHeader<N::BlockHeader>,
        block: &RecoveredBlock<N::Block>,
        output: &BlockExecutionOutput<N::Receipt>,
        trie_updates: Option<(&TrieUpdates, B256)>,
    ) {
        if self.state.invalid_headers.get(&block.hash()).is_some() {
            // we already marked this block as invalid
            return;
        }
        self.invalid_block_hook.on_invalid_block(parent_header, block, output, trie_updates);
    }

    /// This handles downloaded blocks that are shown to be disconnected from the canonical chain.
    ///
    /// This mainly compares the missing parent of the downloaded block with the current canonical
    /// tip, and decides whether or not backfill sync should be triggered.
    fn on_disconnected_downloaded_block(
        &self,
        downloaded_block: BlockNumHash,
        missing_parent: BlockNumHash,
        head: BlockNumHash,
    ) -> Option<TreeEvent> {
        // compare the missing parent with the canonical tip
        if let Some(target) =
            self.backfill_sync_target(head.number, missing_parent.number, Some(downloaded_block))
        {
            trace!(target: "engine::tree", %target, "triggering backfill on downloaded block");
            return Some(TreeEvent::BackfillAction(BackfillAction::Start(target.into())));
        }

        // continue downloading the missing parent
        //
        // this happens if either:
        //  * the missing parent block num < canonical tip num
        //    * this case represents a missing block on a fork that is shorter than the canonical
        //      chain
        //  * the missing parent block num >= canonical tip num, but the number of missing blocks is
        //    less than the backfill threshold
        //    * this case represents a potentially long range of blocks to download and execute
        let request = if let Some(distance) =
            self.distance_from_local_tip(head.number, missing_parent.number)
        {
            trace!(target: "engine::tree", %distance, missing=?missing_parent, "downloading missing parent block range");
            DownloadRequest::BlockRange(missing_parent.hash, distance)
        } else {
            trace!(target: "engine::tree", missing=?missing_parent, "downloading missing parent block");
            // This happens when the missing parent is on an outdated
            // sidechain and we can only download the missing block itself
            DownloadRequest::single_block(missing_parent.hash)
        };

        Some(TreeEvent::Download(request))
    }

    /// Invoked with a block downloaded from the network
    ///
    /// Returns an event with the appropriate action to take, such as:
    ///  - download more missing blocks
    ///  - try to canonicalize the target if the `block` is the tracked target (head) block.
    #[instrument(level = "trace", skip_all, fields(block_hash = %block.hash(), block_num = %block.number(),), target = "engine::tree")]
    fn on_downloaded_block(
        &mut self,
        block: RecoveredBlock<N::Block>,
    ) -> Result<Option<TreeEvent>, InsertBlockFatalError> {
        let block_num_hash = block.num_hash();
        let lowest_buffered_ancestor = self.lowest_buffered_ancestor_or(block_num_hash.hash);
        if self
            .check_invalid_ancestor_with_head(lowest_buffered_ancestor, block.sealed_block())?
            .is_some()
        {
            return Ok(None)
        }

        if !self.backfill_sync_state.is_idle() {
            return Ok(None)
        }

        // try to append the block
        match self.insert_block(block) {
            Ok(InsertPayloadOk::Inserted(BlockStatus::Valid)) => {
                if self.is_sync_target_head(block_num_hash.hash) {
                    trace!(target: "engine::tree", "appended downloaded sync target block");

                    // we just inserted the current sync target block, we can try to make it
                    // canonical
                    return Ok(Some(TreeEvent::TreeAction(TreeAction::MakeCanonical {
                        sync_target_head: block_num_hash.hash,
                    })))
                }
                trace!(target: "engine::tree", "appended downloaded block");
                self.try_connect_buffered_blocks(block_num_hash)?;
            }
            Ok(InsertPayloadOk::Inserted(BlockStatus::Disconnected { head, missing_ancestor })) => {
                // block is not connected to the canonical head, we need to download
                // its missing branch first
                return Ok(self.on_disconnected_downloaded_block(
                    block_num_hash,
                    missing_ancestor,
                    head,
                ))
            }
            Ok(InsertPayloadOk::AlreadySeen(_)) => {
                trace!(target: "engine::tree", "downloaded block already executed");
            }
            Err(err) => {
                debug!(target: "engine::tree", err=%err.kind(), "failed to insert downloaded block");
                if let Err(fatal) = self.on_insert_block_error(err) {
                    warn!(target: "engine::tree", %fatal, "fatal error occurred while inserting downloaded block");
                    return Err(fatal)
                }
            }
        }
        Ok(None)
    }

    fn insert_block(
        &mut self,
        block: RecoveredBlock<N::Block>,
    ) -> Result<InsertPayloadOk, InsertBlockError<N::Block>> {
        self.insert_block_inner(block.clone())
            .map_err(|kind| InsertBlockError::new(block.into_sealed_block(), kind))
    }

    fn insert_block_inner(
        &mut self,
        block: RecoveredBlock<N::Block>,
    ) -> Result<InsertPayloadOk, InsertBlockErrorKind> {
        let block_num_hash = block.num_hash();
        debug!(target: "engine::tree", block=?block_num_hash, parent = ?block.parent_hash(), state_root = ?block.state_root(), "Inserting new block into tree");

        if self.block_by_hash(block.hash())?.is_some() {
            return Ok(InsertPayloadOk::AlreadySeen(BlockStatus::Valid))
        }

        let start = Instant::now();

        trace!(target: "engine::tree", block=?block_num_hash, "Validating block consensus");

        // validate block consensus rules
        self.validate_block(&block)?;

        trace!(target: "engine::tree", block=?block_num_hash, parent=?block.parent_hash(), "Fetching block state provider");
        let Some(provider_builder) = self.state_provider_builder(block.parent_hash())? else {
            // we don't have the state required to execute this block, buffering it and find the
            // missing parent block
            let missing_ancestor = self
                .state
                .buffer
                .lowest_ancestor(&block.parent_hash())
                .map(|block| block.parent_num_hash())
                .unwrap_or_else(|| block.parent_num_hash());

            self.state.buffer.insert_block(block);

            return Ok(InsertPayloadOk::Inserted(BlockStatus::Disconnected {
                head: self.state.tree_state.current_canonical_head,
                missing_ancestor,
            }))
        };

        // now validate against the parent
        let parent_block = self.sealed_header_by_hash(block.parent_hash())?.ok_or_else(|| {
            InsertBlockErrorKind::Provider(ProviderError::HeaderNotFound(
                block.parent_hash().into(),
            ))
        })?;
        if let Err(e) =
            self.consensus.validate_header_against_parent(block.sealed_header(), &parent_block)
        {
            warn!(target: "engine::tree", ?block, "Failed to validate header {} against parent: {e}", block.hash());
            return Err(e.into())
        }

        let state_provider = provider_builder.build()?;

        // We only run the parallel state root if we are not currently persisting any blocks or
        // persisting blocks that are all ancestors of the one we are executing.
        //
        // If we're committing ancestor blocks, then: any trie updates being committed are a subset
        // of the in-memory trie updates collected before fetching reverts. So any diff in
        // reverts (pre vs post commit) is already covered by the in-memory trie updates we
        // collect in `compute_state_root_parallel`.
        //
        // See https://github.com/paradigmxyz/reth/issues/12688 for more details
        let persisting_kind = self.persisting_kind_for(block.header());
        let run_parallel_state_root = persisting_kind.can_run_parallel_state_root();

        // use prewarming background task
        let header = block.clone_sealed_header();
        let txs = block.clone_transactions_recovered().collect();
        let mut handle = if run_parallel_state_root && self.config.use_state_root_task() {
            // use background tasks for state root calc
            let consistent_view = ConsistentDbView::new_with_latest_tip(self.provider.clone())?;

            // Compute trie input
            let trie_input_start = Instant::now();
            let trie_input = self
                .compute_trie_input(
                    persisting_kind,
                    consistent_view.clone(),
                    block.header().parent_hash(),
                )
                .map_err(|e| InsertBlockErrorKind::Other(Box::new(e)))?;

            self.metrics
                .block_validation
                .trie_input_duration
                .record(trie_input_start.elapsed().as_secs_f64());

            self.payload_processor.spawn(header, txs, provider_builder, consistent_view, trie_input)
        } else {
            self.payload_processor.spawn_cache_exclusive(header, txs, provider_builder)
        };

        // Use cached state provider before executing, used in execution after prewarming threads
        // complete
        let state_provider = CachedStateProvider::new_with_caches(
            state_provider,
            handle.caches(),
            handle.cache_metrics(),
        );

        trace!(target: "engine::tree", block=?block_num_hash, "Executing block");

        let executor = self.executor_provider.executor(StateProviderDatabase::new(&state_provider));
        let execution_start = Instant::now();
        let output = self.metrics.executor.execute_metered(
            executor,
            &block,
            Box::new(handle.state_hook()),
        )?;
        let execution_finish = Instant::now();
        let execution_time = execution_finish.duration_since(execution_start);
        trace!(target: "engine::tree", elapsed = ?execution_time, number=?block_num_hash.number, "Executed block");

        // after executing the block we can stop executing transactions
        handle.stop_prewarming_execution();

        if let Err(err) = self.consensus.validate_block_post_execution(&block, &output) {
            // call post-block hook
            self.on_invalid_block(&parent_block, &block, &output, None);
            return Err(err.into())
        }

        let hashed_state = self.provider.hashed_post_state(&output.state);

        if let Err(err) = self
            .payload_validator
            .validate_block_post_execution_with_hashed_state(&hashed_state, &block)
        {
            // call post-block hook
            self.on_invalid_block(&parent_block, &block, &output, None);
            return Err(err.into())
        }

        trace!(target: "engine::tree", block=?block_num_hash, "Calculating block state root");

        let root_time = Instant::now();

        let mut maybe_state_root = None;

        if run_parallel_state_root {
            // if we new payload extends the current canonical change we attempt to use the
            // background task or try to compute it in parallel
            if self.config.use_state_root_task() {
                match handle.state_root() {
                    Ok(StateRootComputeOutcome { state_root, trie_updates }) => {
                        let elapsed = execution_finish.elapsed();
                        info!(target: "engine::tree", ?state_root, ?elapsed, "State root task finished");
                        // we double check the state root here for good measure
                        if state_root == block.header().state_root() {
                            maybe_state_root = Some((state_root, trie_updates, elapsed))
                        } else {
                            warn!(
                                target: "engine::tree",
                                ?state_root,
                                block_state_root = ?block.header().state_root(),
                                "State root task returned incorrect state root"
                            );
                        }
                    }
                    Err(error) => {
                        debug!(target: "engine::tree", %error, "Background parallel state root computation failed");
                    }
                }
            } else {
                match self.compute_state_root_parallel(
                    persisting_kind,
                    block.header().parent_hash(),
                    &hashed_state,
                ) {
                    Ok(result) => {
                        info!(
                            target: "engine::tree",
                            block = ?block_num_hash,
                            regular_state_root = ?result.0,
                            "Regular root task finished"
                        );
                        maybe_state_root = Some((result.0, result.1, root_time.elapsed()));
                    }
                    Err(ParallelStateRootError::Provider(ProviderError::ConsistentView(error))) => {
                        debug!(target: "engine::tree", %error, "Parallel state root computation failed consistency check, falling back");
                    }
                    Err(error) => return Err(InsertBlockErrorKind::Other(Box::new(error))),
                }
            }
        }

        let (state_root, trie_output, root_elapsed) = if let Some(maybe_state_root) =
            maybe_state_root
        {
            maybe_state_root
        } else {
            // fallback is to compute the state root regularly in sync
            debug!(target: "engine::tree", block=?block_num_hash, ?persisting_kind, "Failed to compute state root in parallel");
            let (root, updates) = state_provider.state_root_with_updates(hashed_state.clone())?;
            (root, updates, root_time.elapsed())
        };

        self.metrics.block_validation.record_state_root(&trie_output, root_elapsed.as_secs_f64());
        debug!(target: "engine::tree", ?root_elapsed, block=?block_num_hash, "Calculated state root");

        // ensure state root matches
        if state_root != block.header().state_root() {
            // call post-block hook
            self.on_invalid_block(&parent_block, &block, &output, Some((&trie_output, state_root)));
            return Err(ConsensusError::BodyStateRootDiff(
                GotExpected { got: state_root, expected: block.header().state_root() }.into(),
            )
            .into())
        }

        // terminate prewarming task with good state output
        handle.terminate_caching(Some(output.state.clone()));

        let executed: ExecutedBlockWithTrieUpdates<N> = ExecutedBlockWithTrieUpdates {
            block: ExecutedBlock {
                recovered_block: Arc::new(block),
                execution_output: Arc::new(ExecutionOutcome::from((output, block_num_hash.number))),
                hashed_state: Arc::new(hashed_state),
            },
            trie: Arc::new(trie_output),
        };

        // if the parent is the canonical head, we can insert the block as the pending block
        if self.state.tree_state.canonical_block_hash() == executed.recovered_block().parent_hash()
        {
            debug!(target: "engine::tree", pending=?block_num_hash, "updating pending block");
            self.canonical_in_memory_state.set_pending_block(executed.clone());
        }

        self.state.tree_state.insert_executed(executed.clone());
        self.metrics.engine.executed_blocks.set(self.state.tree_state.block_count() as f64);

        // emit insert event
        let elapsed = start.elapsed();
        let engine_event = if self.is_fork(block_num_hash.hash)? {
            BeaconConsensusEngineEvent::ForkBlockAdded(executed, elapsed)
        } else {
            BeaconConsensusEngineEvent::CanonicalBlockAdded(executed, elapsed)
        };
        self.emit_event(EngineApiEvent::BeaconConsensus(engine_event));

        debug!(target: "engine::tree", block=?block_num_hash, "Finished inserting block");
        Ok(InsertPayloadOk::Inserted(BlockStatus::Valid))
    }

    /// Compute state root for the given hashed post state in parallel.
    ///
    /// # Returns
    ///
    /// Returns `Ok(_)` if computed successfully.
    /// Returns `Err(_)` if error was encountered during computation.
    /// `Err(ProviderError::ConsistentView(_))` can be safely ignored and fallback computation
    /// should be used instead.
    fn compute_state_root_parallel(
        &self,
        persisting_kind: PersistingKind,
        parent_hash: B256,
        hashed_state: &HashedPostState,
    ) -> Result<(B256, TrieUpdates), ParallelStateRootError> {
        let consistent_view = ConsistentDbView::new_with_latest_tip(self.provider.clone())?;

        let mut input =
            self.compute_trie_input(persisting_kind, consistent_view.clone(), parent_hash)?;
        // Extend with block we are validating root for.
        input.append_ref(hashed_state);

        ParallelStateRoot::new(consistent_view, input).incremental_root_with_updates()
    }

    /// Computes the trie input at the provided parent hash.
    ///
    /// The goal of this function is to take in-memory blocks and generate a [`TrieInput`] that
    /// serves as an overlay to the database blocks.
    ///
    /// It works as follows:
    /// 1. Collect in-memory blocks that are descendants of the provided parent hash using
    ///    [`TreeState::blocks_by_hash`].
    /// 2. If the persistence is in progress, and the block that we're computing the trie input for
    ///    is a descendant of the currently persisting blocks, we need to be sure that in-memory
    ///    blocks are not overlapping with the database blocks that may have been already persisted.
    ///    To do that, we're filtering out in-memory blocks that are lower than the highest database
    ///    block.
    /// 3. Once in-memory blocks are collected and optionally filtered, we compute the
    ///    [`HashedPostState`] from them.
    fn compute_trie_input(
        &self,
        persisting_kind: PersistingKind,
        consistent_view: ConsistentDbView<P>,
        parent_hash: B256,
    ) -> Result<TrieInput, ParallelStateRootError> {
        let mut input = TrieInput::default();

        let provider = consistent_view.provider_ro()?;
        let best_block_number = provider.best_block_number()?;

        let (mut historical, mut blocks) = self
            .state
            .tree_state
            .blocks_by_hash(parent_hash)
            .map_or_else(|| (parent_hash.into(), vec![]), |(hash, blocks)| (hash.into(), blocks));

        // If the current block is a descendant of the currently persisting blocks, then we need to
        // filter in-memory blocks, so that none of them are already persisted in the database.
        if persisting_kind.is_descendant() {
            // Iterate over the blocks from oldest to newest.
            while let Some(block) = blocks.last() {
                let recovered_block = block.recovered_block();
                if recovered_block.number() <= best_block_number {
                    // Remove those blocks that lower than or equal to the highest database
                    // block.
                    blocks.pop();
                } else {
                    // If the block is higher than the best block number, stop filtering, as it's
                    // the first block that's not in the database.
                    break
                }
            }

            historical = if let Some(block) = blocks.last() {
                // If there are any in-memory blocks left after filtering, set the anchor to the
                // parent of the oldest block.
                (block.recovered_block().number() - 1).into()
            } else {
                // Otherwise, set the anchor to the original provided parent hash.
                parent_hash.into()
            };
        }

        if blocks.is_empty() {
            debug!(target: "engine::tree", %parent_hash, "Parent found on disk");
        } else {
            debug!(target: "engine::tree", %parent_hash, %historical, blocks = blocks.len(), "Parent found in memory");
        }

        // Convert the historical block to the block number.
        let block_number = provider
            .convert_hash_or_number(historical)?
            .ok_or_else(|| ProviderError::BlockHashNotFound(historical.as_hash().unwrap()))?;

        // Retrieve revert state for historical block.
        let revert_state = if block_number == best_block_number {
            // We do not check against the `last_block_number` here because
            // `HashedPostState::from_reverts` only uses the database tables, and not static files.
            debug!(target: "engine::tree", block_number, best_block_number, "Empty revert state");
            HashedPostState::default()
        } else {
            let revert_state = HashedPostState::from_reverts::<
                <P::StateCommitment as StateCommitment>::KeyHasher,
            >(provider.tx_ref(), block_number + 1)
            .map_err(ProviderError::from)?;
            debug!(
                target: "engine::tree",
                block_number,
                best_block_number,
                accounts = revert_state.accounts.len(),
                storages = revert_state.storages.len(),
                "Non-empty revert state"
            );
            revert_state
        };
        input.append(revert_state);

        // Extend with contents of parent in-memory blocks.
        for block in blocks.iter().rev() {
            input.append_cached_ref(block.trie_updates(), block.hashed_state())
        }

        Ok(input)
    }

    /// Handles an error that occurred while inserting a block.
    ///
    /// If this is a validation error this will mark the block as invalid.
    ///
    /// Returns the proper payload status response if the block is invalid.
    fn on_insert_block_error(
        &mut self,
        error: InsertBlockError<N::Block>,
    ) -> Result<PayloadStatus, InsertBlockFatalError> {
        let (block, error) = error.split();

        // if invalid block, we check the validation error. Otherwise return the fatal
        // error.
        let validation_err = error.ensure_validation_error()?;

        // If the error was due to an invalid payload, the payload is added to the
        // invalid headers cache and `Ok` with [PayloadStatusEnum::Invalid] is
        // returned.
        warn!(
            target: "engine::tree",
            invalid_hash=%block.hash(),
            invalid_number=block.number(),
            %validation_err,
            "Invalid block error on new payload",
        );
        let latest_valid_hash = self.latest_valid_hash_for_invalid_payload(block.parent_hash())?;

        // keep track of the invalid header
        self.state.invalid_headers.insert(block.block_with_parent());
        self.emit_event(EngineApiEvent::BeaconConsensus(BeaconConsensusEngineEvent::InvalidBlock(
            Box::new(block),
        )));
        Ok(PayloadStatus::new(
            PayloadStatusEnum::Invalid { validation_error: validation_err.to_string() },
            latest_valid_hash,
        ))
    }

    /// Attempts to find the header for the given block hash if it is canonical.
    pub fn find_canonical_header(
        &self,
        hash: B256,
    ) -> Result<Option<SealedHeader<N::BlockHeader>>, ProviderError> {
        let mut canonical = self.canonical_in_memory_state.header_by_hash(hash);

        if canonical.is_none() {
            canonical = self.provider.header(&hash)?.map(|header| SealedHeader::new(header, hash));
        }

        Ok(canonical)
    }

    /// Updates the tracked finalized block if we have it.
    fn update_finalized_block(
        &self,
        finalized_block_hash: B256,
    ) -> Result<(), OnForkChoiceUpdated> {
        if finalized_block_hash.is_zero() {
            return Ok(())
        }

        match self.find_canonical_header(finalized_block_hash) {
            Ok(None) => {
                debug!(target: "engine::tree", "Finalized block not found in canonical chain");
                // if the finalized block is not known, we can't update the finalized block
                return Err(OnForkChoiceUpdated::invalid_state())
            }
            Ok(Some(finalized)) => {
                if Some(finalized.num_hash()) !=
                    self.canonical_in_memory_state.get_finalized_num_hash()
                {
                    // we're also persisting the finalized block on disk so we can reload it on
                    // restart this is required by optimism which queries the finalized block: <https://github.com/ethereum-optimism/optimism/blob/c383eb880f307caa3ca41010ec10f30f08396b2e/op-node/rollup/sync/start.go#L65-L65>
                    let _ = self.persistence.save_finalized_block_number(finalized.number());
                    self.canonical_in_memory_state.set_finalized(finalized);
                }
            }
            Err(err) => {
                error!(target: "engine::tree", %err, "Failed to fetch finalized block header");
            }
        }

        Ok(())
    }

    /// Updates the tracked safe block if we have it
    fn update_safe_block(&self, safe_block_hash: B256) -> Result<(), OnForkChoiceUpdated> {
        if safe_block_hash.is_zero() {
            return Ok(())
        }

        match self.find_canonical_header(safe_block_hash) {
            Ok(None) => {
                debug!(target: "engine::tree", "Safe block not found in canonical chain");
                // if the safe block is not known, we can't update the safe block
                return Err(OnForkChoiceUpdated::invalid_state())
            }
            Ok(Some(safe)) => {
                if Some(safe.num_hash()) != self.canonical_in_memory_state.get_safe_num_hash() {
                    // we're also persisting the safe block on disk so we can reload it on
                    // restart this is required by optimism which queries the safe block: <https://github.com/ethereum-optimism/optimism/blob/c383eb880f307caa3ca41010ec10f30f08396b2e/op-node/rollup/sync/start.go#L65-L65>
                    let _ = self.persistence.save_safe_block_number(safe.number());
                    self.canonical_in_memory_state.set_safe(safe);
                }
            }
            Err(err) => {
                error!(target: "engine::tree", %err, "Failed to fetch safe block header");
            }
        }

        Ok(())
    }

    /// Ensures that the given forkchoice state is consistent, assuming the head block has been
    /// made canonical.
    ///
    /// If the forkchoice state is consistent, this will return Ok(()). Otherwise, this will
    /// return an instance of [`OnForkChoiceUpdated`] that is INVALID.
    ///
    /// This also updates the safe and finalized blocks in the [`CanonicalInMemoryState`], if they
    /// are consistent with the head block.
    fn ensure_consistent_forkchoice_state(
        &self,
        state: ForkchoiceState,
    ) -> Result<(), OnForkChoiceUpdated> {
        // Ensure that the finalized block, if not zero, is known and in the canonical chain
        // after the head block is canonicalized.
        //
        // This ensures that the finalized block is consistent with the head block, i.e. the
        // finalized block is an ancestor of the head block.
        self.update_finalized_block(state.finalized_block_hash)?;

        // Also ensure that the safe block, if not zero, is known and in the canonical chain
        // after the head block is canonicalized.
        //
        // This ensures that the safe block is consistent with the head block, i.e. the safe
        // block is an ancestor of the head block.
        self.update_safe_block(state.safe_block_hash)
    }

    /// Pre-validate forkchoice update and check whether it can be processed.
    ///
    /// This method returns the update outcome if validation fails or
    /// the node is syncing and the update cannot be processed at the moment.
    fn pre_validate_forkchoice_update(
        &mut self,
        state: ForkchoiceState,
    ) -> ProviderResult<Option<OnForkChoiceUpdated>> {
        if state.head_block_hash.is_zero() {
            return Ok(Some(OnForkChoiceUpdated::invalid_state()))
        }

        // check if the new head hash is connected to any ancestor that we previously marked as
        // invalid
        let lowest_buffered_ancestor_fcu = self.lowest_buffered_ancestor_or(state.head_block_hash);
        if let Some(status) = self.check_invalid_ancestor(lowest_buffered_ancestor_fcu)? {
            return Ok(Some(OnForkChoiceUpdated::with_invalid(status)))
        }

        if !self.backfill_sync_state.is_idle() {
            // We can only process new forkchoice updates if the pipeline is idle, since it requires
            // exclusive access to the database
            trace!(target: "engine::tree", "Pipeline is syncing, skipping forkchoice update");
            return Ok(Some(OnForkChoiceUpdated::syncing()))
        }

        Ok(None)
    }

    /// Validates the payload attributes with respect to the header and fork choice state.
    ///
    /// Note: At this point, the fork choice update is considered to be VALID, however, we can still
    /// return an error if the payload attributes are invalid.
    fn process_payload_attributes(
        &self,
        attrs: T::PayloadAttributes,
        head: &N::BlockHeader,
        state: ForkchoiceState,
        version: EngineApiMessageVersion,
    ) -> OnForkChoiceUpdated {
        if let Err(err) =
            self.payload_validator.validate_payload_attributes_against_header(&attrs, head)
        {
            warn!(target: "engine::tree", %err, ?head, "Invalid payload attributes");
            return OnForkChoiceUpdated::invalid_payload_attributes()
        }

        // 8. Client software MUST begin a payload build process building on top of
        //    forkchoiceState.headBlockHash and identified via buildProcessId value if
        //    payloadAttributes is not null and the forkchoice state has been updated successfully.
        //    The build process is specified in the Payload building section.
        match <T::PayloadBuilderAttributes as PayloadBuilderAttributes>::try_new(
            state.head_block_hash,
            attrs,
            version as u8,
        ) {
            Ok(attributes) => {
                // send the payload to the builder and return the receiver for the pending payload
                // id, initiating payload job is handled asynchronously
                let pending_payload_id = self.payload_builder.send_new_payload(attributes);

                // Client software MUST respond to this method call in the following way:
                // {
                //      payloadStatus: {
                //          status: VALID,
                //          latestValidHash: forkchoiceState.headBlockHash,
                //          validationError: null
                //      },
                //      payloadId: buildProcessId
                // }
                //
                // if the payload is deemed VALID and the build process has begun.
                OnForkChoiceUpdated::updated_with_pending_payload_id(
                    PayloadStatus::new(PayloadStatusEnum::Valid, Some(state.head_block_hash)),
                    pending_payload_id,
                )
            }
            Err(_) => OnForkChoiceUpdated::invalid_payload_attributes(),
        }
    }

    /// Remove all blocks up to __and including__ the given block number.
    ///
    /// If a finalized hash is provided, the only non-canonical blocks which will be removed are
    /// those which have a fork point at or below the finalized hash.
    ///
    /// Canonical blocks below the upper bound will still be removed.
    pub(crate) fn remove_before(
        &mut self,
        upper_bound: BlockNumHash,
        finalized_hash: Option<B256>,
    ) -> ProviderResult<()> {
        // first fetch the finalized block number and then call the remove_before method on
        // tree_state
        let num = if let Some(hash) = finalized_hash {
            self.provider.block_number(hash)?.map(|number| BlockNumHash { number, hash })
        } else {
            None
        };

        self.state.tree_state.remove_until(
            upper_bound,
            self.persistence_state.last_persisted_block.hash,
            num,
        );
        Ok(())
    }

    /// Returns a builder for creating state providers for the given hash.
    ///
    /// This is an optimization for parallel execution contexts where we want to avoid
    /// creating state providers in the critical path.
    pub fn state_provider_builder(
        &self,
        hash: B256,
    ) -> ProviderResult<Option<StateProviderBuilder<N, P>>>
    where
        P: BlockReader + StateProviderFactory + StateReader + StateCommitmentProvider + Clone,
    {
        if let Some((historical, blocks)) = self.state.tree_state.blocks_by_hash(hash) {
            debug!(target: "engine::tree", %hash, %historical, "found canonical state for block in memory, creating provider builder");
            // the block leads back to the canonical chain
            return Ok(Some(StateProviderBuilder::new(
                self.provider.clone(),
                historical,
                Some(blocks),
            )))
        }

        // Check if the block is persisted
        if let Some(header) = self.provider.header(&hash)? {
            debug!(target: "engine::tree", %hash, number = %header.number(), "found canonical state for block in database, creating provider builder");
            // For persisted blocks, we create a builder that will fetch state directly from the
            // database
            return Ok(Some(StateProviderBuilder::new(self.provider.clone(), hash, None)))
        }

        debug!(target: "engine::tree", %hash, "no canonical state found for block");
        Ok(None)
    }
}

/// Block inclusion can be valid, accepted, or invalid. Invalid blocks are returned as an error
/// variant.
///
/// If we don't know the block's parent, we return `Disconnected`, as we can't claim that the block
/// is valid or not.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum BlockStatus {
    /// The block is valid and block extends canonical chain.
    Valid,
    /// The block may be valid and has an unknown missing ancestor.
    Disconnected {
        /// Current canonical head.
        head: BlockNumHash,
        /// The lowest ancestor block that is not connected to the canonical chain.
        missing_ancestor: BlockNumHash,
    },
}

/// How a payload was inserted if it was valid.
///
/// If the payload was valid, but has already been seen, [`InsertPayloadOk::AlreadySeen(_)`] is
/// returned, otherwise [`InsertPayloadOk::Inserted(_)`] is returned.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum InsertPayloadOk {
    /// The payload was valid, but we have already seen it.
    AlreadySeen(BlockStatus),
    /// The payload was valid and inserted into the tree.
    Inserted(BlockStatus),
}

/// Whether or not the blocks are currently persisting and the input block is a descendant.
#[derive(Debug, Clone, Copy)]
pub enum PersistingKind {
    /// The blocks are not currently persisting.
    NotPersisting,
    /// The blocks are currently persisting but the input block is not a descendant.
    PersistingNotDescendant,
    /// The blocks are currently persisting and the input block is a descendant.
    PersistingDescendant,
}

impl PersistingKind {
    /// Returns true if the parallel state root can be run.
    ///
    /// We only run the parallel state root if we are not currently persisting any blocks or
    /// persisting blocks that are all ancestors of the one we are calculating the state root for.
    pub fn can_run_parallel_state_root(&self) -> bool {
        matches!(self, Self::NotPersisting | Self::PersistingDescendant)
    }

    /// Returns true if the blocks are currently being persisted and the input block is a
    /// descendant.
    pub fn is_descendant(&self) -> bool {
        matches!(self, Self::PersistingDescendant)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::persistence::PersistenceAction;
    use alloy_consensus::Header;
    use alloy_primitives::Bytes;
    use alloy_rlp::Decodable;
    use alloy_rpc_types_engine::{
        CancunPayloadFields, ExecutionData, ExecutionPayloadSidecar, ExecutionPayloadV1,
        ExecutionPayloadV3,
    };
    use assert_matches::assert_matches;
    use reth_chain_state::{test_utils::TestBlockBuilder, BlockState};
    use reth_chainspec::{ChainSpec, HOLESKY, MAINNET};
    use reth_engine_primitives::ForkchoiceStatus;
    use reth_ethereum_consensus::EthBeaconConsensus;
    use reth_ethereum_engine_primitives::EthEngineTypes;
    use reth_ethereum_primitives::{Block, EthPrimitives};
    use reth_evm::test_utils::MockExecutorProvider;
    use reth_evm_ethereum::EthEvmConfig;
    use reth_node_ethereum::EthereumEngineValidator;
    use reth_primitives_traits::Block as _;
    use reth_provider::test_utils::MockEthProvider;
    use reth_trie::{updates::TrieUpdates, HashedPostState};
    use std::{
        str::FromStr,
        sync::mpsc::{channel, Sender},
    };

    /// This is a test channel that allows you to `release` any value that is in the channel.
    ///
    /// If nothing has been sent, then the next value will be immediately sent.
    #[allow(dead_code)]
    struct TestChannel<T> {
        /// If an item is sent to this channel, an item will be released in the wrapped channel
        release: Receiver<()>,
        /// The sender channel
        tx: Sender<T>,
        /// The receiver channel
        rx: Receiver<T>,
    }

    impl<T: Send + 'static> TestChannel<T> {
        /// Creates a new test channel
        #[allow(dead_code)]
        fn spawn_channel() -> (Sender<T>, Receiver<T>, TestChannelHandle) {
            let (original_tx, original_rx) = channel();
            let (wrapped_tx, wrapped_rx) = channel();
            let (release_tx, release_rx) = channel();
            let handle = TestChannelHandle::new(release_tx);
            let test_channel = Self { release: release_rx, tx: wrapped_tx, rx: original_rx };
            // spawn the task that listens and releases stuff
            std::thread::spawn(move || test_channel.intercept_loop());
            (original_tx, wrapped_rx, handle)
        }

        /// Runs the intercept loop, waiting for the handle to release a value
        fn intercept_loop(&self) {
            while self.release.recv() == Ok(()) {
                let Ok(value) = self.rx.recv() else { return };

                let _ = self.tx.send(value);
            }
        }
    }

    struct TestChannelHandle {
        /// The sender to use for releasing values
        release: Sender<()>,
    }

    impl TestChannelHandle {
        /// Returns a [`TestChannelHandle`]
        const fn new(release: Sender<()>) -> Self {
            Self { release }
        }

        /// Signals to the channel task that a value should be released
        #[allow(dead_code)]
        fn release(&self) {
            let _ = self.release.send(());
        }
    }

    struct TestHarness {
        tree: EngineApiTreeHandler<
            EthPrimitives,
            MockEthProvider,
            MockExecutorProvider,
            EthEngineTypes,
            EthereumEngineValidator,
            EthEvmConfig,
        >,
        to_tree_tx: Sender<FromEngine<EngineApiRequest<EthEngineTypes, EthPrimitives>, Block>>,
        from_tree_rx: UnboundedReceiver<EngineApiEvent>,
        blocks: Vec<ExecutedBlockWithTrieUpdates>,
        action_rx: Receiver<PersistenceAction>,
        executor_provider: MockExecutorProvider,
        block_builder: TestBlockBuilder,
        provider: MockEthProvider,
    }

    impl TestHarness {
        fn new(chain_spec: Arc<ChainSpec>) -> Self {
            let (action_tx, action_rx) = channel();
            Self::with_persistence_channel(chain_spec, action_tx, action_rx)
        }

        #[allow(dead_code)]
        fn with_test_channel(chain_spec: Arc<ChainSpec>) -> (Self, TestChannelHandle) {
            let (action_tx, action_rx, handle) = TestChannel::spawn_channel();
            (Self::with_persistence_channel(chain_spec, action_tx, action_rx), handle)
        }

        fn with_persistence_channel(
            chain_spec: Arc<ChainSpec>,
            action_tx: Sender<PersistenceAction>,
            action_rx: Receiver<PersistenceAction>,
        ) -> Self {
            let persistence_handle = PersistenceHandle::new(action_tx);

            let consensus = Arc::new(EthBeaconConsensus::new(chain_spec.clone()));

            let provider = MockEthProvider::default();
            let executor_provider = MockExecutorProvider::default();

            let payload_validator = EthereumEngineValidator::new(chain_spec.clone());

            let (from_tree_tx, from_tree_rx) = unbounded_channel();

            let header = chain_spec.genesis_header().clone();
            let header = SealedHeader::seal_slow(header);
            let engine_api_tree_state = EngineApiTreeState::new(10, 10, header.num_hash());
            let canonical_in_memory_state = CanonicalInMemoryState::with_head(header, None, None);

            let (to_payload_service, _payload_command_rx) = unbounded_channel();
            let payload_builder = PayloadBuilderHandle::new(to_payload_service);

            let evm_config = EthEvmConfig::new(chain_spec.clone());

            let tree = EngineApiTreeHandler::new(
                provider.clone(),
                executor_provider.clone(),
                consensus,
                payload_validator,
                from_tree_tx,
                engine_api_tree_state,
                canonical_in_memory_state,
                persistence_handle,
                PersistenceState::default(),
                payload_builder,
                // TODO: fix tests for state root task https://github.com/paradigmxyz/reth/issues/14376
                // always assume enough parallelism for tests
                TreeConfig::default()
                    .with_legacy_state_root(true)
                    .with_has_enough_parallelism(true),
                EngineApiKind::Ethereum,
                evm_config,
            );

            let block_builder = TestBlockBuilder::default().with_chain_spec((*chain_spec).clone());
            Self {
                to_tree_tx: tree.incoming_tx.clone(),
                tree,
                from_tree_rx,
                blocks: vec![],
                action_rx,
                executor_provider,
                block_builder,
                provider,
            }
        }

        fn with_blocks(mut self, blocks: Vec<ExecutedBlockWithTrieUpdates>) -> Self {
            let mut blocks_by_hash = HashMap::default();
            let mut blocks_by_number = BTreeMap::new();
            let mut state_by_hash = HashMap::default();
            let mut hash_by_number = BTreeMap::new();
            let mut parent_to_child: HashMap<B256, HashSet<B256>> = HashMap::default();
            let mut parent_hash = B256::ZERO;

            for block in &blocks {
                let sealed_block = block.recovered_block();
                let hash = sealed_block.hash();
                let number = sealed_block.number;
                blocks_by_hash.insert(hash, block.clone());
                blocks_by_number.entry(number).or_insert_with(Vec::new).push(block.clone());
                state_by_hash.insert(hash, Arc::new(BlockState::new(block.clone())));
                hash_by_number.insert(number, hash);
                parent_to_child.entry(parent_hash).or_default().insert(hash);
                parent_hash = hash;
            }

            self.tree.state.tree_state = TreeState {
                blocks_by_hash,
                blocks_by_number,
                current_canonical_head: blocks.last().unwrap().recovered_block().num_hash(),
                parent_to_child,
                persisted_trie_updates: HashMap::default(),
            };

            let last_executed_block = blocks.last().unwrap().clone();
            let pending = Some(BlockState::new(last_executed_block));
            self.tree.canonical_in_memory_state =
                CanonicalInMemoryState::new(state_by_hash, hash_by_number, pending, None, None);

            self.blocks = blocks.clone();

            let recovered_blocks =
                blocks.iter().map(|b| b.recovered_block().clone()).collect::<Vec<_>>();

            self.persist_blocks(recovered_blocks);

            self
        }

        const fn with_backfill_state(mut self, state: BackfillSyncState) -> Self {
            self.tree.backfill_sync_state = state;
            self
        }

        fn extend_execution_outcome(
            &self,
            execution_outcomes: impl IntoIterator<Item = impl Into<ExecutionOutcome>>,
        ) {
            self.executor_provider.extend(execution_outcomes);
        }

        fn insert_block(
            &mut self,
            block: RecoveredBlock<reth_ethereum_primitives::Block>,
        ) -> Result<InsertPayloadOk, InsertBlockError<Block>> {
            let execution_outcome = self.block_builder.get_execution_outcome(block.clone());
            self.extend_execution_outcome([execution_outcome]);
            self.tree.provider.add_state_root(block.state_root);
            self.tree.insert_block(block)
        }

        async fn fcu_to(&mut self, block_hash: B256, fcu_status: impl Into<ForkchoiceStatus>) {
            let fcu_status = fcu_status.into();

            self.send_fcu(block_hash, fcu_status).await;

            self.check_fcu(block_hash, fcu_status).await;
        }

        async fn send_fcu(&mut self, block_hash: B256, fcu_status: impl Into<ForkchoiceStatus>) {
            let fcu_state = self.fcu_state(block_hash);

            let (tx, rx) = oneshot::channel();
            self.tree
                .on_engine_message(FromEngine::Request(
                    BeaconEngineMessage::ForkchoiceUpdated {
                        state: fcu_state,
                        payload_attrs: None,
                        tx,
                        version: EngineApiMessageVersion::default(),
                    }
                    .into(),
                ))
                .unwrap();

            let response = rx.await.unwrap().unwrap().await.unwrap();
            match fcu_status.into() {
                ForkchoiceStatus::Valid => assert!(response.payload_status.is_valid()),
                ForkchoiceStatus::Syncing => assert!(response.payload_status.is_syncing()),
                ForkchoiceStatus::Invalid => assert!(response.payload_status.is_invalid()),
            }
        }

        async fn check_fcu(&mut self, block_hash: B256, fcu_status: impl Into<ForkchoiceStatus>) {
            let fcu_state = self.fcu_state(block_hash);

            // check for ForkchoiceUpdated event
            let event = self.from_tree_rx.recv().await.unwrap();
            match event {
                EngineApiEvent::BeaconConsensus(BeaconConsensusEngineEvent::ForkchoiceUpdated(
                    state,
                    status,
                )) => {
                    assert_eq!(state, fcu_state);
                    assert_eq!(status, fcu_status.into());
                }
                _ => panic!("Unexpected event: {:#?}", event),
            }
        }

        const fn fcu_state(&self, block_hash: B256) -> ForkchoiceState {
            ForkchoiceState {
                head_block_hash: block_hash,
                safe_block_hash: block_hash,
                finalized_block_hash: block_hash,
            }
        }

        async fn send_new_payload(
            &mut self,
            block: RecoveredBlock<reth_ethereum_primitives::Block>,
        ) {
            let payload = ExecutionPayloadV3::from_block_unchecked(
                block.hash(),
                &block.clone_sealed_block().into_block(),
            );
            self.tree
                .on_new_payload(ExecutionData {
                    payload: payload.into(),
                    sidecar: ExecutionPayloadSidecar::v3(CancunPayloadFields {
                        parent_beacon_block_root: block.parent_beacon_block_root.unwrap(),
                        versioned_hashes: vec![],
                    }),
                })
                .unwrap();
        }

        async fn insert_chain(
            &mut self,
            chain: impl IntoIterator<Item = RecoveredBlock<reth_ethereum_primitives::Block>> + Clone,
        ) {
            for block in chain.clone() {
                self.insert_block(block.clone()).unwrap();
            }
            self.check_canon_chain_insertion(chain).await;
        }

        async fn check_canon_commit(&mut self, hash: B256) {
            let event = self.from_tree_rx.recv().await.unwrap();
            match event {
                EngineApiEvent::BeaconConsensus(
                    BeaconConsensusEngineEvent::CanonicalChainCommitted(header, _),
                ) => {
                    assert_eq!(header.hash(), hash);
                }
                _ => panic!("Unexpected event: {:#?}", event),
            }
        }

        async fn check_fork_chain_insertion(
            &mut self,
            chain: impl IntoIterator<Item = RecoveredBlock<reth_ethereum_primitives::Block>> + Clone,
        ) {
            for block in chain {
                self.check_fork_block_added(block.hash()).await;
            }
        }

        async fn check_canon_chain_insertion(
            &mut self,
            chain: impl IntoIterator<Item = RecoveredBlock<reth_ethereum_primitives::Block>> + Clone,
        ) {
            for block in chain.clone() {
                self.check_canon_block_added(block.hash()).await;
            }
        }

        async fn check_canon_block_added(&mut self, expected_hash: B256) {
            let event = self.from_tree_rx.recv().await.unwrap();
            match event {
                EngineApiEvent::BeaconConsensus(
                    BeaconConsensusEngineEvent::CanonicalBlockAdded(executed, _),
                ) => {
                    assert_eq!(executed.recovered_block.hash(), expected_hash);
                }
                _ => panic!("Unexpected event: {:#?}", event),
            }
        }

        async fn check_fork_block_added(&mut self, expected_hash: B256) {
            let event = self.from_tree_rx.recv().await.unwrap();
            match event {
                EngineApiEvent::BeaconConsensus(BeaconConsensusEngineEvent::ForkBlockAdded(
                    executed,
                    _,
                )) => {
                    assert_eq!(executed.recovered_block.hash(), expected_hash);
                }
                _ => panic!("Unexpected event: {:#?}", event),
            }
        }

        async fn check_invalid_block(&mut self, expected_hash: B256) {
            let event = self.from_tree_rx.recv().await.unwrap();
            match event {
                EngineApiEvent::BeaconConsensus(BeaconConsensusEngineEvent::InvalidBlock(
                    block,
                )) => {
                    assert_eq!(block.hash(), expected_hash);
                }
                _ => panic!("Unexpected event: {:#?}", event),
            }
        }

        fn persist_blocks(&self, blocks: Vec<RecoveredBlock<reth_ethereum_primitives::Block>>) {
            let mut block_data: Vec<(B256, Block)> = Vec::with_capacity(blocks.len());
            let mut headers_data: Vec<(B256, Header)> = Vec::with_capacity(blocks.len());

            for block in &blocks {
                block_data.push((block.hash(), block.clone_block()));
                headers_data.push((block.hash(), block.header().clone()));
            }

            self.provider.extend_blocks(block_data);
            self.provider.extend_headers(headers_data);
        }

        fn setup_range_insertion_for_valid_chain(
            &mut self,
            chain: Vec<RecoveredBlock<reth_ethereum_primitives::Block>>,
        ) {
            self.setup_range_insertion_for_chain(chain, None)
        }

        fn setup_range_insertion_for_invalid_chain(
            &mut self,
            chain: Vec<RecoveredBlock<reth_ethereum_primitives::Block>>,
            index: usize,
        ) {
            self.setup_range_insertion_for_chain(chain, Some(index))
        }

        fn setup_range_insertion_for_chain(
            &mut self,
            chain: Vec<RecoveredBlock<reth_ethereum_primitives::Block>>,
            invalid_index: Option<usize>,
        ) {
            // setting up execution outcomes for the chain, the blocks will be
            // executed starting from the oldest, so we need to reverse.
            let mut chain_rev = chain;
            chain_rev.reverse();

            let mut execution_outcomes = Vec::with_capacity(chain_rev.len());
            for (index, block) in chain_rev.iter().enumerate() {
                let execution_outcome = self.block_builder.get_execution_outcome(block.clone());
                let state_root = if invalid_index.is_some() && invalid_index.unwrap() == index {
                    B256::random()
                } else {
                    block.state_root
                };
                self.tree.provider.add_state_root(state_root);
                execution_outcomes.push(execution_outcome);
            }
            self.extend_execution_outcome(execution_outcomes);
        }

        fn check_canon_head(&self, head_hash: B256) {
            assert_eq!(self.tree.state.tree_state.canonical_head().hash, head_hash);
        }
    }

    #[test]
    fn test_tree_persist_block_batch() {
        let tree_config = TreeConfig::default();
        let chain_spec = MAINNET.clone();
        let mut test_block_builder = TestBlockBuilder::eth().with_chain_spec((*chain_spec).clone());

        // we need more than tree_config.persistence_threshold() +1 blocks to
        // trigger the persistence task.
        let blocks: Vec<_> = test_block_builder
            .get_executed_blocks(1..tree_config.persistence_threshold() + 2)
            .collect();
        let mut test_harness = TestHarness::new(chain_spec).with_blocks(blocks);

        let mut blocks = vec![];
        for idx in 0..tree_config.max_execute_block_batch_size() * 2 {
            blocks.push(test_block_builder.generate_random_block(idx as u64, B256::random()));
        }

        test_harness.to_tree_tx.send(FromEngine::DownloadedBlocks(blocks)).unwrap();

        // process the message
        let msg = test_harness.tree.try_recv_engine_message().unwrap().unwrap();
        test_harness.tree.on_engine_message(msg).unwrap();

        // we now should receive the other batch
        let msg = test_harness.tree.try_recv_engine_message().unwrap().unwrap();
        match msg {
            FromEngine::DownloadedBlocks(blocks) => {
                assert_eq!(blocks.len(), tree_config.max_execute_block_batch_size());
            }
            _ => panic!("unexpected message: {:#?}", msg),
        }
    }

    #[tokio::test]
    async fn test_tree_persist_blocks() {
        let tree_config = TreeConfig::default();
        let chain_spec = MAINNET.clone();
        let mut test_block_builder = TestBlockBuilder::eth().with_chain_spec((*chain_spec).clone());

        // we need more than tree_config.persistence_threshold() +1 blocks to
        // trigger the persistence task.
        let blocks: Vec<_> = test_block_builder
            .get_executed_blocks(1..tree_config.persistence_threshold() + 2)
            .collect();
        let test_harness = TestHarness::new(chain_spec).with_blocks(blocks.clone());
        std::thread::Builder::new()
            .name("Tree Task".to_string())
            .spawn(|| test_harness.tree.run())
            .unwrap();

        // send a message to the tree to enter the main loop.
        test_harness.to_tree_tx.send(FromEngine::DownloadedBlocks(vec![])).unwrap();

        let received_action =
            test_harness.action_rx.recv().expect("Failed to receive save blocks action");
        if let PersistenceAction::SaveBlocks(saved_blocks, _) = received_action {
            // only blocks.len() - tree_config.memory_block_buffer_target() will be
            // persisted
            let expected_persist_len =
                blocks.len() - tree_config.memory_block_buffer_target() as usize;
            assert_eq!(saved_blocks.len(), expected_persist_len);
            assert_eq!(saved_blocks, blocks[..expected_persist_len]);
        } else {
            panic!("unexpected action received {received_action:?}");
        }
    }

    #[tokio::test]
    async fn test_in_memory_state_trait_impl() {
        let blocks: Vec<_> = TestBlockBuilder::eth().get_executed_blocks(0..10).collect();
        let test_harness = TestHarness::new(MAINNET.clone()).with_blocks(blocks.clone());

        for executed_block in blocks {
            let sealed_block = executed_block.recovered_block();

            let expected_state = BlockState::new(executed_block.clone());

            let actual_state_by_hash = test_harness
                .tree
                .canonical_in_memory_state
                .state_by_hash(sealed_block.hash())
                .unwrap();
            assert_eq!(expected_state, *actual_state_by_hash);

            let actual_state_by_number = test_harness
                .tree
                .canonical_in_memory_state
                .state_by_number(sealed_block.number)
                .unwrap();
            assert_eq!(expected_state, *actual_state_by_number);
        }
    }

    #[tokio::test]
    async fn test_engine_request_during_backfill() {
        let tree_config = TreeConfig::default();
        let blocks: Vec<_> = TestBlockBuilder::eth()
            .get_executed_blocks(0..tree_config.persistence_threshold())
            .collect();
        let mut test_harness = TestHarness::new(MAINNET.clone())
            .with_blocks(blocks)
            .with_backfill_state(BackfillSyncState::Active);

        let (tx, rx) = oneshot::channel();
        test_harness
            .tree
            .on_engine_message(FromEngine::Request(
                BeaconEngineMessage::ForkchoiceUpdated {
                    state: ForkchoiceState {
                        head_block_hash: B256::random(),
                        safe_block_hash: B256::random(),
                        finalized_block_hash: B256::random(),
                    },
                    payload_attrs: None,
                    tx,
                    version: EngineApiMessageVersion::default(),
                }
                .into(),
            ))
            .unwrap();

        let resp = rx.await.unwrap().unwrap().await.unwrap();
        assert!(resp.payload_status.is_syncing());
    }

    #[test]
    fn test_disconnected_payload() {
        let s = include_str!("../../test-data/holesky/2.rlp");
        let data = Bytes::from_str(s).unwrap();
        let block = Block::decode(&mut data.as_ref()).unwrap();
        let sealed = block.seal_slow();
        let hash = sealed.hash();
        let payload = ExecutionPayloadV1::from_block_unchecked(hash, &sealed.clone().into_block());

        let mut test_harness = TestHarness::new(HOLESKY.clone());

        let outcome = test_harness
            .tree
            .on_new_payload(ExecutionData {
                payload: payload.into(),
                sidecar: ExecutionPayloadSidecar::none(),
            })
            .unwrap();
        assert!(outcome.outcome.is_syncing());

        // ensure block is buffered
        let buffered = test_harness.tree.state.buffer.block(&hash).unwrap();
        assert_eq!(buffered.clone_sealed_block(), sealed);
    }

    #[test]
    fn test_disconnected_block() {
        let s = include_str!("../../test-data/holesky/2.rlp");
        let data = Bytes::from_str(s).unwrap();
        let block = Block::decode(&mut data.as_ref()).unwrap();
        let sealed = block.seal_slow().try_recover().unwrap();

        let mut test_harness = TestHarness::new(HOLESKY.clone());

        let outcome = test_harness.tree.insert_block(sealed.clone()).unwrap();
        assert_eq!(
            outcome,
            InsertPayloadOk::Inserted(BlockStatus::Disconnected {
                head: test_harness.tree.state.tree_state.current_canonical_head,
                missing_ancestor: sealed.parent_num_hash()
            })
        );
    }

    #[tokio::test]
    async fn test_holesky_payload() {
        let s = include_str!("../../test-data/holesky/1.rlp");
        let data = Bytes::from_str(s).unwrap();
        let block: Block = Block::decode(&mut data.as_ref()).unwrap();
        let sealed = block.seal_slow();
        let payload =
            ExecutionPayloadV1::from_block_unchecked(sealed.hash(), &sealed.clone().into_block());

        let mut test_harness =
            TestHarness::new(HOLESKY.clone()).with_backfill_state(BackfillSyncState::Active);

        let (tx, rx) = oneshot::channel();
        test_harness
            .tree
            .on_engine_message(FromEngine::Request(
                BeaconEngineMessage::NewPayload {
                    payload: ExecutionData {
                        payload: payload.clone().into(),
                        sidecar: ExecutionPayloadSidecar::none(),
                    },
                    tx,
                }
                .into(),
            ))
            .unwrap();

        let resp = rx.await.unwrap().unwrap();
        assert!(resp.is_syncing());
    }

    #[test]
    fn test_tree_state_normal_descendant() {
        let mut tree_state = TreeState::new(BlockNumHash::default());
        let blocks: Vec<_> = TestBlockBuilder::eth().get_executed_blocks(1..4).collect();

        tree_state.insert_executed(blocks[0].clone());
        assert!(tree_state.is_descendant(
            blocks[0].recovered_block().num_hash(),
            blocks[1].recovered_block().header()
        ));

        tree_state.insert_executed(blocks[1].clone());

        assert!(tree_state.is_descendant(
            blocks[0].recovered_block().num_hash(),
            blocks[2].recovered_block().header()
        ));
        assert!(tree_state.is_descendant(
            blocks[1].recovered_block().num_hash(),
            blocks[2].recovered_block().header()
        ));
    }

    #[tokio::test]
    async fn test_tree_state_insert_executed() {
        let mut tree_state = TreeState::new(BlockNumHash::default());
        let blocks: Vec<_> = TestBlockBuilder::eth().get_executed_blocks(1..4).collect();

        tree_state.insert_executed(blocks[0].clone());
        tree_state.insert_executed(blocks[1].clone());

        assert_eq!(
            tree_state.parent_to_child.get(&blocks[0].recovered_block().hash()),
            Some(&HashSet::from_iter([blocks[1].recovered_block().hash()]))
        );

        assert!(!tree_state.parent_to_child.contains_key(&blocks[1].recovered_block().hash()));

        tree_state.insert_executed(blocks[2].clone());

        assert_eq!(
            tree_state.parent_to_child.get(&blocks[1].recovered_block().hash()),
            Some(&HashSet::from_iter([blocks[2].recovered_block().hash()]))
        );
        assert!(tree_state.parent_to_child.contains_key(&blocks[1].recovered_block().hash()));

        assert!(!tree_state.parent_to_child.contains_key(&blocks[2].recovered_block().hash()));
    }

    #[tokio::test]
    async fn test_tree_state_insert_executed_with_reorg() {
        let mut tree_state = TreeState::new(BlockNumHash::default());
        let mut test_block_builder = TestBlockBuilder::eth();
        let blocks: Vec<_> = test_block_builder.get_executed_blocks(1..6).collect();

        for block in &blocks {
            tree_state.insert_executed(block.clone());
        }
        assert_eq!(tree_state.blocks_by_hash.len(), 5);

        let fork_block_3 = test_block_builder
            .get_executed_block_with_number(3, blocks[1].recovered_block().hash());
        let fork_block_4 = test_block_builder
            .get_executed_block_with_number(4, fork_block_3.recovered_block().hash());
        let fork_block_5 = test_block_builder
            .get_executed_block_with_number(5, fork_block_4.recovered_block().hash());

        tree_state.insert_executed(fork_block_3.clone());
        tree_state.insert_executed(fork_block_4.clone());
        tree_state.insert_executed(fork_block_5.clone());

        assert_eq!(tree_state.blocks_by_hash.len(), 8);
        assert_eq!(tree_state.blocks_by_number[&3].len(), 2); // two blocks at height 3 (original and fork)
        assert_eq!(tree_state.parent_to_child[&blocks[1].recovered_block().hash()].len(), 2); // block 2 should have two children

        // verify that we can insert the same block again without issues
        tree_state.insert_executed(fork_block_4.clone());
        assert_eq!(tree_state.blocks_by_hash.len(), 8);

        assert!(tree_state.parent_to_child[&fork_block_3.recovered_block().hash()]
            .contains(&fork_block_4.recovered_block().hash()));
        assert!(tree_state.parent_to_child[&fork_block_4.recovered_block().hash()]
            .contains(&fork_block_5.recovered_block().hash()));

        assert_eq!(tree_state.blocks_by_number[&4].len(), 2);
        assert_eq!(tree_state.blocks_by_number[&5].len(), 2);
    }

    #[tokio::test]
    async fn test_tree_state_remove_before() {
        let start_num_hash = BlockNumHash::default();
        let mut tree_state = TreeState::new(start_num_hash);
        let blocks: Vec<_> = TestBlockBuilder::eth().get_executed_blocks(1..6).collect();

        for block in &blocks {
            tree_state.insert_executed(block.clone());
        }

        let last = blocks.last().unwrap();

        // set the canonical head
        tree_state.set_canonical_head(last.recovered_block().num_hash());

        // inclusive bound, so we should remove anything up to and including 2
        tree_state.remove_until(
            BlockNumHash::new(2, blocks[1].recovered_block().hash()),
            start_num_hash.hash,
            Some(blocks[1].recovered_block().num_hash()),
        );

        assert!(!tree_state.blocks_by_hash.contains_key(&blocks[0].recovered_block().hash()));
        assert!(!tree_state.blocks_by_hash.contains_key(&blocks[1].recovered_block().hash()));
        assert!(!tree_state.blocks_by_number.contains_key(&1));
        assert!(!tree_state.blocks_by_number.contains_key(&2));

        assert!(tree_state.blocks_by_hash.contains_key(&blocks[2].recovered_block().hash()));
        assert!(tree_state.blocks_by_hash.contains_key(&blocks[3].recovered_block().hash()));
        assert!(tree_state.blocks_by_hash.contains_key(&blocks[4].recovered_block().hash()));
        assert!(tree_state.blocks_by_number.contains_key(&3));
        assert!(tree_state.blocks_by_number.contains_key(&4));
        assert!(tree_state.blocks_by_number.contains_key(&5));

        assert!(!tree_state.parent_to_child.contains_key(&blocks[0].recovered_block().hash()));
        assert!(!tree_state.parent_to_child.contains_key(&blocks[1].recovered_block().hash()));
        assert!(tree_state.parent_to_child.contains_key(&blocks[2].recovered_block().hash()));
        assert!(tree_state.parent_to_child.contains_key(&blocks[3].recovered_block().hash()));
        assert!(!tree_state.parent_to_child.contains_key(&blocks[4].recovered_block().hash()));

        assert_eq!(
            tree_state.parent_to_child.get(&blocks[2].recovered_block().hash()),
            Some(&HashSet::from_iter([blocks[3].recovered_block().hash()]))
        );
        assert_eq!(
            tree_state.parent_to_child.get(&blocks[3].recovered_block().hash()),
            Some(&HashSet::from_iter([blocks[4].recovered_block().hash()]))
        );
    }

    #[tokio::test]
    async fn test_tree_state_remove_before_finalized() {
        let start_num_hash = BlockNumHash::default();
        let mut tree_state = TreeState::new(start_num_hash);
        let blocks: Vec<_> = TestBlockBuilder::eth().get_executed_blocks(1..6).collect();

        for block in &blocks {
            tree_state.insert_executed(block.clone());
        }

        let last = blocks.last().unwrap();

        // set the canonical head
        tree_state.set_canonical_head(last.recovered_block().num_hash());

        // we should still remove everything up to and including 2
        tree_state.remove_until(
            BlockNumHash::new(2, blocks[1].recovered_block().hash()),
            start_num_hash.hash,
            None,
        );

        assert!(!tree_state.blocks_by_hash.contains_key(&blocks[0].recovered_block().hash()));
        assert!(!tree_state.blocks_by_hash.contains_key(&blocks[1].recovered_block().hash()));
        assert!(!tree_state.blocks_by_number.contains_key(&1));
        assert!(!tree_state.blocks_by_number.contains_key(&2));

        assert!(tree_state.blocks_by_hash.contains_key(&blocks[2].recovered_block().hash()));
        assert!(tree_state.blocks_by_hash.contains_key(&blocks[3].recovered_block().hash()));
        assert!(tree_state.blocks_by_hash.contains_key(&blocks[4].recovered_block().hash()));
        assert!(tree_state.blocks_by_number.contains_key(&3));
        assert!(tree_state.blocks_by_number.contains_key(&4));
        assert!(tree_state.blocks_by_number.contains_key(&5));

        assert!(!tree_state.parent_to_child.contains_key(&blocks[0].recovered_block().hash()));
        assert!(!tree_state.parent_to_child.contains_key(&blocks[1].recovered_block().hash()));
        assert!(tree_state.parent_to_child.contains_key(&blocks[2].recovered_block().hash()));
        assert!(tree_state.parent_to_child.contains_key(&blocks[3].recovered_block().hash()));
        assert!(!tree_state.parent_to_child.contains_key(&blocks[4].recovered_block().hash()));

        assert_eq!(
            tree_state.parent_to_child.get(&blocks[2].recovered_block().hash()),
            Some(&HashSet::from_iter([blocks[3].recovered_block().hash()]))
        );
        assert_eq!(
            tree_state.parent_to_child.get(&blocks[3].recovered_block().hash()),
            Some(&HashSet::from_iter([blocks[4].recovered_block().hash()]))
        );
    }

    #[tokio::test]
    async fn test_tree_state_remove_before_lower_finalized() {
        let start_num_hash = BlockNumHash::default();
        let mut tree_state = TreeState::new(start_num_hash);
        let blocks: Vec<_> = TestBlockBuilder::eth().get_executed_blocks(1..6).collect();

        for block in &blocks {
            tree_state.insert_executed(block.clone());
        }

        let last = blocks.last().unwrap();

        // set the canonical head
        tree_state.set_canonical_head(last.recovered_block().num_hash());

        // we have no forks so we should still remove anything up to and including 2
        tree_state.remove_until(
            BlockNumHash::new(2, blocks[1].recovered_block().hash()),
            start_num_hash.hash,
            Some(blocks[0].recovered_block().num_hash()),
        );

        assert!(!tree_state.blocks_by_hash.contains_key(&blocks[0].recovered_block().hash()));
        assert!(!tree_state.blocks_by_hash.contains_key(&blocks[1].recovered_block().hash()));
        assert!(!tree_state.blocks_by_number.contains_key(&1));
        assert!(!tree_state.blocks_by_number.contains_key(&2));

        assert!(tree_state.blocks_by_hash.contains_key(&blocks[2].recovered_block().hash()));
        assert!(tree_state.blocks_by_hash.contains_key(&blocks[3].recovered_block().hash()));
        assert!(tree_state.blocks_by_hash.contains_key(&blocks[4].recovered_block().hash()));
        assert!(tree_state.blocks_by_number.contains_key(&3));
        assert!(tree_state.blocks_by_number.contains_key(&4));
        assert!(tree_state.blocks_by_number.contains_key(&5));

        assert!(!tree_state.parent_to_child.contains_key(&blocks[0].recovered_block().hash()));
        assert!(!tree_state.parent_to_child.contains_key(&blocks[1].recovered_block().hash()));
        assert!(tree_state.parent_to_child.contains_key(&blocks[2].recovered_block().hash()));
        assert!(tree_state.parent_to_child.contains_key(&blocks[3].recovered_block().hash()));
        assert!(!tree_state.parent_to_child.contains_key(&blocks[4].recovered_block().hash()));

        assert_eq!(
            tree_state.parent_to_child.get(&blocks[2].recovered_block().hash()),
            Some(&HashSet::from_iter([blocks[3].recovered_block().hash()]))
        );
        assert_eq!(
            tree_state.parent_to_child.get(&blocks[3].recovered_block().hash()),
            Some(&HashSet::from_iter([blocks[4].recovered_block().hash()]))
        );
    }

    #[tokio::test]
    async fn test_tree_state_on_new_head_reorg() {
        reth_tracing::init_test_tracing();
        let chain_spec = MAINNET.clone();

        // Set persistence_threshold to 1
        let mut test_harness = TestHarness::new(chain_spec);
        test_harness.tree.config = test_harness
            .tree
            .config
            .with_persistence_threshold(1)
            .with_memory_block_buffer_target(1);
        let mut test_block_builder = TestBlockBuilder::eth();
        let blocks: Vec<_> = test_block_builder.get_executed_blocks(1..6).collect();

        for block in &blocks {
            test_harness.tree.state.tree_state.insert_executed(block.clone());
        }

        // set block 3 as the current canonical head
        test_harness
            .tree
            .state
            .tree_state
            .set_canonical_head(blocks[2].recovered_block().num_hash());

        // create a fork from block 2
        let fork_block_3 = test_block_builder
            .get_executed_block_with_number(3, blocks[1].recovered_block().hash());
        let fork_block_4 = test_block_builder
            .get_executed_block_with_number(4, fork_block_3.recovered_block().hash());
        let fork_block_5 = test_block_builder
            .get_executed_block_with_number(5, fork_block_4.recovered_block().hash());

        test_harness.tree.state.tree_state.insert_executed(fork_block_3.clone());
        test_harness.tree.state.tree_state.insert_executed(fork_block_4.clone());
        test_harness.tree.state.tree_state.insert_executed(fork_block_5.clone());

        // normal (non-reorg) case
        let result = test_harness.tree.on_new_head(blocks[4].recovered_block().hash()).unwrap();
        assert!(matches!(result, Some(NewCanonicalChain::Commit { .. })));
        if let Some(NewCanonicalChain::Commit { new }) = result {
            assert_eq!(new.len(), 2);
            assert_eq!(new[0].recovered_block().hash(), blocks[3].recovered_block().hash());
            assert_eq!(new[1].recovered_block().hash(), blocks[4].recovered_block().hash());
        }

        // should be a None persistence action before we advance persistence
        let current_action = test_harness.tree.persistence_state.current_action();
        assert_eq!(current_action, None);

        // let's attempt to persist and check that it attempts to save blocks
        //
        // since in-memory block buffer target and persistence_threshold are both 1, this should
        // save all but the current tip of the canonical chain (up to blocks[1])
        test_harness.tree.advance_persistence().unwrap();
        let current_action = test_harness.tree.persistence_state.current_action().cloned();
        assert_eq!(
            current_action,
            Some(CurrentPersistenceAction::SavingBlocks {
                highest: blocks[1].recovered_block().num_hash()
            })
        );

        // get rid of the prev action
        let received_action = test_harness.action_rx.recv().unwrap();
        let PersistenceAction::SaveBlocks(saved_blocks, sender) = received_action else {
            panic!("received wrong action");
        };
        assert_eq!(saved_blocks, vec![blocks[0].clone(), blocks[1].clone()]);

        // send the response so we can advance again
        sender.send(Some(blocks[1].recovered_block().num_hash())).unwrap();

        // we should be persisting blocks[1] because we threw out the prev action
        let current_action = test_harness.tree.persistence_state.current_action().cloned();
        assert_eq!(
            current_action,
            Some(CurrentPersistenceAction::SavingBlocks {
                highest: blocks[1].recovered_block().num_hash()
            })
        );

        // after advancing persistence, we should be at `None` for the next action
        test_harness.tree.advance_persistence().unwrap();
        let current_action = test_harness.tree.persistence_state.current_action().cloned();
        assert_eq!(current_action, None);

        // reorg case
        let result = test_harness.tree.on_new_head(fork_block_5.recovered_block().hash()).unwrap();
        assert!(matches!(result, Some(NewCanonicalChain::Reorg { .. })));

        if let Some(NewCanonicalChain::Reorg { new, old }) = result {
            assert_eq!(new.len(), 3);
            assert_eq!(new[0].recovered_block().hash(), fork_block_3.recovered_block().hash());
            assert_eq!(new[1].recovered_block().hash(), fork_block_4.recovered_block().hash());
            assert_eq!(new[2].recovered_block().hash(), fork_block_5.recovered_block().hash());

            assert_eq!(old.len(), 1);
            assert_eq!(old[0].recovered_block().hash(), blocks[2].recovered_block().hash());
        }

        // The canonical block has not changed, so we will not get any active persistence action
        test_harness.tree.advance_persistence().unwrap();
        let current_action = test_harness.tree.persistence_state.current_action().cloned();
        assert_eq!(current_action, None);

        // Let's change the canonical head and advance persistence
        test_harness
            .tree
            .state
            .tree_state
            .set_canonical_head(fork_block_5.recovered_block().num_hash());

        // The canonical block has changed now, we should get fork_block_4 due to the persistence
        // threshold and in memory block buffer target
        test_harness.tree.advance_persistence().unwrap();
        let current_action = test_harness.tree.persistence_state.current_action().cloned();
        assert_eq!(
            current_action,
            Some(CurrentPersistenceAction::SavingBlocks {
                highest: fork_block_4.recovered_block().num_hash()
            })
        );
    }

    #[test]
    fn test_tree_state_on_new_head_deep_fork() {
        reth_tracing::init_test_tracing();

        let chain_spec = MAINNET.clone();
        let mut test_harness = TestHarness::new(chain_spec);
        let mut test_block_builder = TestBlockBuilder::eth();

        let blocks: Vec<_> = test_block_builder.get_executed_blocks(0..5).collect();

        for block in &blocks {
            test_harness.tree.state.tree_state.insert_executed(block.clone());
        }

        // set last block as the current canonical head
        let last_block = blocks.last().unwrap().recovered_block().clone();

        test_harness.tree.state.tree_state.set_canonical_head(last_block.num_hash());

        // create a fork chain from last_block
        let chain_a = test_block_builder.create_fork(&last_block, 10);
        let chain_b = test_block_builder.create_fork(&last_block, 10);

        for block in &chain_a {
            test_harness.tree.state.tree_state.insert_executed(ExecutedBlockWithTrieUpdates {
                block: ExecutedBlock {
                    recovered_block: Arc::new(block.clone()),
                    execution_output: Arc::new(ExecutionOutcome::default()),
                    hashed_state: Arc::new(HashedPostState::default()),
                },
                trie: Arc::new(TrieUpdates::default()),
            });
        }
        test_harness.tree.state.tree_state.set_canonical_head(chain_a.last().unwrap().num_hash());

        for block in &chain_b {
            test_harness.tree.state.tree_state.insert_executed(ExecutedBlockWithTrieUpdates {
                block: ExecutedBlock {
                    recovered_block: Arc::new(block.clone()),
                    execution_output: Arc::new(ExecutionOutcome::default()),
                    hashed_state: Arc::new(HashedPostState::default()),
                },
                trie: Arc::new(TrieUpdates::default()),
            });
        }

        // for each block in chain_b, reorg to it and then back to canonical
        let mut expected_new = Vec::new();
        for block in &chain_b {
            // reorg to chain from block b
            let result = test_harness.tree.on_new_head(block.hash()).unwrap();
            assert_matches!(result, Some(NewCanonicalChain::Reorg { .. }));

            expected_new.push(block);
            if let Some(NewCanonicalChain::Reorg { new, old }) = result {
                assert_eq!(new.len(), expected_new.len());
                for (index, block) in expected_new.iter().enumerate() {
                    assert_eq!(new[index].recovered_block().hash(), block.hash());
                }

                assert_eq!(old.len(), chain_a.len());
                for (index, block) in chain_a.iter().enumerate() {
                    assert_eq!(old[index].recovered_block().hash(), block.hash());
                }
            }

            // set last block of chain a as canonical head
            test_harness.tree.on_new_head(chain_a.last().unwrap().hash()).unwrap();
        }
    }

    #[tokio::test]
    async fn test_get_canonical_blocks_to_persist() {
        let chain_spec = MAINNET.clone();
        let mut test_harness = TestHarness::new(chain_spec);
        let mut test_block_builder = TestBlockBuilder::eth();

        let canonical_head_number = 9;
        let blocks: Vec<_> =
            test_block_builder.get_executed_blocks(0..canonical_head_number + 1).collect();
        test_harness = test_harness.with_blocks(blocks.clone());

        let last_persisted_block_number = 3;
        test_harness.tree.persistence_state.last_persisted_block.number =
            last_persisted_block_number;

        let persistence_threshold = 4;
        let memory_block_buffer_target = 3;
        test_harness.tree.config = TreeConfig::default()
            .with_persistence_threshold(persistence_threshold)
            .with_memory_block_buffer_target(memory_block_buffer_target);

        let blocks_to_persist = test_harness.tree.get_canonical_blocks_to_persist();

        let expected_blocks_to_persist_length: usize =
            (canonical_head_number - memory_block_buffer_target - last_persisted_block_number)
                .try_into()
                .unwrap();

        assert_eq!(blocks_to_persist.len(), expected_blocks_to_persist_length);
        for (i, item) in
            blocks_to_persist.iter().enumerate().take(expected_blocks_to_persist_length)
        {
            assert_eq!(item.recovered_block().number, last_persisted_block_number + i as u64 + 1);
        }

        // make sure only canonical blocks are included
        let fork_block = test_block_builder.get_executed_block_with_number(4, B256::random());
        let fork_block_hash = fork_block.recovered_block().hash();
        test_harness.tree.state.tree_state.insert_executed(fork_block);

        assert!(test_harness.tree.state.tree_state.block_by_hash(fork_block_hash).is_some());

        let blocks_to_persist = test_harness.tree.get_canonical_blocks_to_persist();
        assert_eq!(blocks_to_persist.len(), expected_blocks_to_persist_length);

        // check that the fork block is not included in the blocks to persist
        assert!(!blocks_to_persist.iter().any(|b| b.recovered_block().hash() == fork_block_hash));

        // check that the original block 4 is still included
        assert!(blocks_to_persist.iter().any(|b| b.recovered_block().number == 4 &&
            b.recovered_block().hash() == blocks[4].recovered_block().hash()));

        // check that if we advance persistence, the persistence action is the correct value
        test_harness.tree.advance_persistence().expect("advancing persistence should succeed");
        assert_eq!(
            test_harness.tree.persistence_state.current_action().cloned(),
            Some(CurrentPersistenceAction::SavingBlocks {
                highest: blocks_to_persist.last().unwrap().recovered_block().num_hash()
            })
        );
    }

    #[tokio::test]
    async fn test_engine_tree_fcu_missing_head() {
        let chain_spec = MAINNET.clone();
        let mut test_harness = TestHarness::new(chain_spec.clone());

        let mut test_block_builder = TestBlockBuilder::eth().with_chain_spec((*chain_spec).clone());

        let blocks: Vec<_> = test_block_builder.get_executed_blocks(0..5).collect();
        test_harness = test_harness.with_blocks(blocks);

        let missing_block = test_block_builder
            .generate_random_block(6, test_harness.blocks.last().unwrap().recovered_block().hash());

        test_harness.fcu_to(missing_block.hash(), PayloadStatusEnum::Syncing).await;

        // after FCU we receive an EngineApiEvent::Download event to get the missing block.
        let event = test_harness.from_tree_rx.recv().await.unwrap();
        match event {
            EngineApiEvent::Download(DownloadRequest::BlockSet(actual_block_set)) => {
                let expected_block_set = HashSet::from_iter([missing_block.hash()]);
                assert_eq!(actual_block_set, expected_block_set);
            }
            _ => panic!("Unexpected event: {:#?}", event),
        }
    }

    #[tokio::test]
    async fn test_engine_tree_fcu_canon_chain_insertion() {
        let chain_spec = MAINNET.clone();
        let mut test_harness = TestHarness::new(chain_spec.clone());

        let base_chain: Vec<_> = test_harness.block_builder.get_executed_blocks(0..1).collect();
        test_harness = test_harness.with_blocks(base_chain.clone());

        test_harness
            .fcu_to(base_chain.last().unwrap().recovered_block().hash(), ForkchoiceStatus::Valid)
            .await;

        // extend main chain
        let main_chain = test_harness.block_builder.create_fork(base_chain[0].recovered_block(), 3);

        test_harness.insert_chain(main_chain).await;
    }

    #[tokio::test]
    async fn test_engine_tree_fcu_reorg_with_all_blocks() {
        let chain_spec = MAINNET.clone();
        let mut test_harness = TestHarness::new(chain_spec.clone());

        let main_chain: Vec<_> = test_harness.block_builder.get_executed_blocks(0..5).collect();
        test_harness = test_harness.with_blocks(main_chain.clone());

        let fork_chain = test_harness.block_builder.create_fork(main_chain[2].recovered_block(), 3);
        let fork_chain_last_hash = fork_chain.last().unwrap().hash();

        // add fork blocks to the tree
        for block in &fork_chain {
            test_harness.insert_block(block.clone()).unwrap();
        }

        test_harness.send_fcu(fork_chain_last_hash, ForkchoiceStatus::Valid).await;

        // check for ForkBlockAdded events, we expect fork_chain.len() blocks added
        test_harness.check_fork_chain_insertion(fork_chain.clone()).await;

        // check for CanonicalChainCommitted event
        test_harness.check_canon_commit(fork_chain_last_hash).await;

        test_harness.check_fcu(fork_chain_last_hash, ForkchoiceStatus::Valid).await;

        // new head is the tip of the fork chain
        test_harness.check_canon_head(fork_chain_last_hash);
    }

    #[tokio::test]
    async fn test_engine_tree_live_sync_transition_required_blocks_requested() {
        reth_tracing::init_test_tracing();

        let chain_spec = MAINNET.clone();
        let mut test_harness = TestHarness::new(chain_spec.clone());

        let base_chain: Vec<_> = test_harness.block_builder.get_executed_blocks(0..1).collect();
        test_harness = test_harness.with_blocks(base_chain.clone());

        test_harness
            .fcu_to(base_chain.last().unwrap().recovered_block().hash(), ForkchoiceStatus::Valid)
            .await;

        // extend main chain with enough blocks to trigger pipeline run but don't insert them
        let main_chain = test_harness
            .block_builder
            .create_fork(base_chain[0].recovered_block(), MIN_BLOCKS_FOR_PIPELINE_RUN + 10);

        let main_chain_last_hash = main_chain.last().unwrap().hash();
        test_harness.send_fcu(main_chain_last_hash, ForkchoiceStatus::Syncing).await;

        test_harness.check_fcu(main_chain_last_hash, ForkchoiceStatus::Syncing).await;

        // create event for backfill finished
        let backfill_finished_block_number = MIN_BLOCKS_FOR_PIPELINE_RUN + 1;
        let backfill_finished = FromOrchestrator::BackfillSyncFinished(ControlFlow::Continue {
            block_number: backfill_finished_block_number,
        });

        let backfill_tip_block = main_chain[(backfill_finished_block_number - 1) as usize].clone();
        // add block to mock provider to enable persistence clean up.
        test_harness.provider.add_block(backfill_tip_block.hash(), backfill_tip_block.into_block());
        test_harness.tree.on_engine_message(FromEngine::Event(backfill_finished)).unwrap();

        let event = test_harness.from_tree_rx.recv().await.unwrap();
        match event {
            EngineApiEvent::Download(DownloadRequest::BlockSet(hash_set)) => {
                assert_eq!(hash_set, HashSet::from_iter([main_chain_last_hash]));
            }
            _ => panic!("Unexpected event: {:#?}", event),
        }

        test_harness
            .tree
            .on_engine_message(FromEngine::DownloadedBlocks(vec![main_chain
                .last()
                .unwrap()
                .clone()]))
            .unwrap();

        let event = test_harness.from_tree_rx.recv().await.unwrap();
        match event {
            EngineApiEvent::Download(DownloadRequest::BlockRange(initial_hash, total_blocks)) => {
                assert_eq!(
                    total_blocks,
                    (main_chain.len() - backfill_finished_block_number as usize - 1) as u64
                );
                assert_eq!(initial_hash, main_chain.last().unwrap().parent_hash);
            }
            _ => panic!("Unexpected event: {:#?}", event),
        }
    }

    #[tokio::test]
    async fn test_engine_tree_live_sync_transition_eventually_canonical() {
        reth_tracing::init_test_tracing();

        let chain_spec = MAINNET.clone();
        let mut test_harness = TestHarness::new(chain_spec.clone());
        test_harness.tree.config = test_harness.tree.config.with_max_execute_block_batch_size(100);

        // create base chain and setup test harness with it
        let base_chain: Vec<_> = test_harness.block_builder.get_executed_blocks(0..1).collect();
        test_harness = test_harness.with_blocks(base_chain.clone());

        // fcu to the tip of base chain
        test_harness
            .fcu_to(base_chain.last().unwrap().recovered_block().hash(), ForkchoiceStatus::Valid)
            .await;

        // create main chain, extension of base chain, with enough blocks to
        // trigger backfill sync
        let main_chain = test_harness
            .block_builder
            .create_fork(base_chain[0].recovered_block(), MIN_BLOCKS_FOR_PIPELINE_RUN + 10);

        let main_chain_last = main_chain.last().unwrap();
        let main_chain_last_hash = main_chain_last.hash();
        let main_chain_backfill_target =
            main_chain.get(MIN_BLOCKS_FOR_PIPELINE_RUN as usize).unwrap();
        let main_chain_backfill_target_hash = main_chain_backfill_target.hash();

        // fcu to the element of main chain that should trigger backfill sync
        test_harness.send_fcu(main_chain_backfill_target_hash, ForkchoiceStatus::Syncing).await;
        test_harness.check_fcu(main_chain_backfill_target_hash, ForkchoiceStatus::Syncing).await;

        // check download request for target
        let event = test_harness.from_tree_rx.recv().await.unwrap();
        match event {
            EngineApiEvent::Download(DownloadRequest::BlockSet(hash_set)) => {
                assert_eq!(hash_set, HashSet::from_iter([main_chain_backfill_target_hash]));
            }
            _ => panic!("Unexpected event: {:#?}", event),
        }

        // send message to tell the engine the requested block was downloaded
        test_harness
            .tree
            .on_engine_message(FromEngine::DownloadedBlocks(vec![
                main_chain_backfill_target.clone()
            ]))
            .unwrap();

        // check that backfill is triggered
        let event = test_harness.from_tree_rx.recv().await.unwrap();
        match event {
            EngineApiEvent::BackfillAction(BackfillAction::Start(
                reth_stages::PipelineTarget::Sync(target_hash),
            )) => {
                assert_eq!(target_hash, main_chain_backfill_target_hash);
            }
            _ => panic!("Unexpected event: {:#?}", event),
        }

        // persist blocks of main chain, same as the backfill operation would do
        let backfilled_chain: Vec<_> =
            main_chain.clone().drain(0..(MIN_BLOCKS_FOR_PIPELINE_RUN + 1) as usize).collect();
        test_harness.persist_blocks(backfilled_chain.clone());

        test_harness.setup_range_insertion_for_valid_chain(backfilled_chain);

        // send message to mark backfill finished
        test_harness
            .tree
            .on_engine_message(FromEngine::Event(FromOrchestrator::BackfillSyncFinished(
                ControlFlow::Continue { block_number: main_chain_backfill_target.number },
            )))
            .unwrap();

        // send fcu to the tip of main
        test_harness.fcu_to(main_chain_last_hash, ForkchoiceStatus::Syncing).await;

        let event = test_harness.from_tree_rx.recv().await.unwrap();
        match event {
            EngineApiEvent::Download(DownloadRequest::BlockSet(target_hash)) => {
                assert_eq!(target_hash, HashSet::from_iter([main_chain_last_hash]));
            }
            _ => panic!("Unexpected event: {:#?}", event),
        }

        // tell engine main chain tip downloaded
        test_harness
            .tree
            .on_engine_message(FromEngine::DownloadedBlocks(vec![main_chain_last.clone()]))
            .unwrap();

        // check download range request
        let event = test_harness.from_tree_rx.recv().await.unwrap();
        match event {
            EngineApiEvent::Download(DownloadRequest::BlockRange(initial_hash, total_blocks)) => {
                assert_eq!(
                    total_blocks,
                    (main_chain.len() - MIN_BLOCKS_FOR_PIPELINE_RUN as usize - 2) as u64
                );
                assert_eq!(initial_hash, main_chain_last.parent_hash);
            }
            _ => panic!("Unexpected event: {:#?}", event),
        }

        let remaining: Vec<_> = main_chain
            .clone()
            .drain((MIN_BLOCKS_FOR_PIPELINE_RUN + 1) as usize..main_chain.len())
            .collect();

        test_harness.setup_range_insertion_for_valid_chain(remaining.clone());

        // tell engine block range downloaded
        test_harness
            .tree
            .on_engine_message(FromEngine::DownloadedBlocks(remaining.clone()))
            .unwrap();

        test_harness.check_canon_chain_insertion(remaining).await;

        // check canonical chain committed event with the hash of the latest block
        test_harness.check_canon_commit(main_chain_last_hash).await;

        // new head is the tip of the main chain
        test_harness.check_canon_head(main_chain_last_hash);
    }

    #[tokio::test]
    async fn test_engine_tree_live_sync_fcu_extends_canon_chain() {
        reth_tracing::init_test_tracing();

        let chain_spec = MAINNET.clone();
        let mut test_harness = TestHarness::new(chain_spec.clone());

        // create base chain and setup test harness with it
        let base_chain: Vec<_> = test_harness.block_builder.get_executed_blocks(0..1).collect();
        test_harness = test_harness.with_blocks(base_chain.clone());

        // fcu to the tip of base chain
        test_harness
            .fcu_to(base_chain.last().unwrap().recovered_block().hash(), ForkchoiceStatus::Valid)
            .await;

        // create main chain, extension of base chain
        let main_chain =
            test_harness.block_builder.create_fork(base_chain[0].recovered_block(), 10);
        // determine target in the middle of main hain
        let target = main_chain.get(5).unwrap();
        let target_hash = target.hash();
        let main_last = main_chain.last().unwrap();
        let main_last_hash = main_last.hash();

        // insert main chain
        test_harness.insert_chain(main_chain).await;

        // send fcu to target
        test_harness.send_fcu(target_hash, ForkchoiceStatus::Valid).await;

        test_harness.check_canon_commit(target_hash).await;
        test_harness.check_fcu(target_hash, ForkchoiceStatus::Valid).await;

        // send fcu to main tip
        test_harness.send_fcu(main_last_hash, ForkchoiceStatus::Valid).await;

        test_harness.check_canon_commit(main_last_hash).await;
        test_harness.check_fcu(main_last_hash, ForkchoiceStatus::Valid).await;
        test_harness.check_canon_head(main_last_hash);
    }

    #[tokio::test]
    async fn test_engine_tree_valid_forks_with_older_canonical_head() {
        reth_tracing::init_test_tracing();

        let chain_spec = MAINNET.clone();
        let mut test_harness = TestHarness::new(chain_spec.clone());

        // create base chain and setup test harness with it
        let base_chain: Vec<_> = test_harness.block_builder.get_executed_blocks(0..1).collect();
        test_harness = test_harness.with_blocks(base_chain.clone());

        let old_head = base_chain.first().unwrap().recovered_block();

        // extend base chain
        let extension_chain = test_harness.block_builder.create_fork(old_head, 5);
        let fork_block = extension_chain.last().unwrap().clone_sealed_block();

        test_harness.setup_range_insertion_for_valid_chain(extension_chain.clone());
        test_harness.insert_chain(extension_chain).await;

        // fcu to old_head
        test_harness.fcu_to(old_head.hash(), ForkchoiceStatus::Valid).await;

        // create two competing chains starting from fork_block
        let chain_a = test_harness.block_builder.create_fork(&fork_block, 10);
        let chain_b = test_harness.block_builder.create_fork(&fork_block, 10);

        // insert chain A blocks using newPayload
        test_harness.setup_range_insertion_for_valid_chain(chain_a.clone());
        for block in &chain_a {
            test_harness.send_new_payload(block.clone()).await;
        }

        test_harness.check_canon_chain_insertion(chain_a.clone()).await;

        // insert chain B blocks using newPayload
        test_harness.setup_range_insertion_for_valid_chain(chain_b.clone());
        for block in &chain_b {
            test_harness.send_new_payload(block.clone()).await;
        }

        test_harness.check_canon_chain_insertion(chain_b.clone()).await;

        // send FCU to make the tip of chain B the new head
        let chain_b_tip_hash = chain_b.last().unwrap().hash();
        test_harness.send_fcu(chain_b_tip_hash, ForkchoiceStatus::Valid).await;

        // check for CanonicalChainCommitted event
        test_harness.check_canon_commit(chain_b_tip_hash).await;

        // verify FCU was processed
        test_harness.check_fcu(chain_b_tip_hash, ForkchoiceStatus::Valid).await;

        // verify the new canonical head
        test_harness.check_canon_head(chain_b_tip_hash);

        // verify that chain A is now considered a fork
        assert!(test_harness.tree.is_fork(chain_a.last().unwrap().hash()).unwrap());
    }

    #[tokio::test]
    async fn test_engine_tree_buffered_blocks_are_eventually_connected() {
        let chain_spec = MAINNET.clone();
        let mut test_harness = TestHarness::new(chain_spec.clone());

        let base_chain: Vec<_> = test_harness.block_builder.get_executed_blocks(0..1).collect();
        test_harness = test_harness.with_blocks(base_chain.clone());

        // side chain consisting of two blocks, the last will be inserted first
        // so that we force it to be buffered
        let side_chain =
            test_harness.block_builder.create_fork(base_chain.last().unwrap().recovered_block(), 2);

        // buffer last block of side chain
        let buffered_block = side_chain.last().unwrap();
        let buffered_block_hash = buffered_block.hash();

        test_harness.setup_range_insertion_for_valid_chain(vec![buffered_block.clone()]);
        test_harness.send_new_payload(buffered_block.clone()).await;

        assert!(test_harness.tree.state.buffer.block(&buffered_block_hash).is_some());

        let non_buffered_block = side_chain.first().unwrap();
        let non_buffered_block_hash = non_buffered_block.hash();

        // insert block that continues the canon chain, should not be buffered
        test_harness.setup_range_insertion_for_valid_chain(vec![non_buffered_block.clone()]);
        test_harness.send_new_payload(non_buffered_block.clone()).await;
        assert!(test_harness.tree.state.buffer.block(&non_buffered_block_hash).is_none());

        // the previously buffered block should be connected now
        assert!(test_harness.tree.state.buffer.block(&buffered_block_hash).is_none());

        // both blocks are added to the canon chain in order
        test_harness.check_canon_block_added(non_buffered_block_hash).await;
        test_harness.check_canon_block_added(buffered_block_hash).await;
    }

    #[tokio::test]
    async fn test_engine_tree_valid_and_invalid_forks_with_older_canonical_head() {
        reth_tracing::init_test_tracing();

        let chain_spec = MAINNET.clone();
        let mut test_harness = TestHarness::new(chain_spec.clone());

        // create base chain and setup test harness with it
        let base_chain: Vec<_> = test_harness.block_builder.get_executed_blocks(0..1).collect();
        test_harness = test_harness.with_blocks(base_chain.clone());

        let old_head = base_chain.first().unwrap().recovered_block();

        // extend base chain
        let extension_chain = test_harness.block_builder.create_fork(old_head, 5);
        let fork_block = extension_chain.last().unwrap().clone_sealed_block();
        test_harness.insert_chain(extension_chain).await;

        // fcu to old_head
        test_harness.fcu_to(old_head.hash(), ForkchoiceStatus::Valid).await;

        // create two competing chains starting from fork_block, one of them invalid
        let total_fork_elements = 10;
        let chain_a = test_harness.block_builder.create_fork(&fork_block, total_fork_elements);
        let chain_b = test_harness.block_builder.create_fork(&fork_block, total_fork_elements);

        // insert chain B blocks using newPayload
        test_harness.setup_range_insertion_for_valid_chain(chain_b.clone());
        for block in &chain_b {
            test_harness.send_new_payload(block.clone()).await;
            test_harness.send_fcu(block.hash(), ForkchoiceStatus::Valid).await;
            test_harness.check_canon_block_added(block.hash()).await;
            test_harness.check_canon_commit(block.hash()).await;
            test_harness.check_fcu(block.hash(), ForkchoiceStatus::Valid).await;
        }

        // insert chain A blocks using newPayload, one of the blocks will be invalid
        let invalid_index = 3;
        test_harness.setup_range_insertion_for_invalid_chain(chain_a.clone(), invalid_index);
        for block in &chain_a {
            test_harness.send_new_payload(block.clone()).await;
        }

        // check canon chain insertion up to the invalid index and taking into
        // account reversed ordering
        test_harness
            .check_fork_chain_insertion(
                chain_a[..chain_a.len() - invalid_index - 1].iter().cloned(),
            )
            .await;
        for block in &chain_a[chain_a.len() - invalid_index - 1..] {
            test_harness.check_invalid_block(block.hash()).await;
        }

        // send FCU to make the tip of chain A, expect invalid
        let chain_a_tip_hash = chain_a.last().unwrap().hash();
        test_harness.fcu_to(chain_a_tip_hash, ForkchoiceStatus::Invalid).await;

        // send FCU to make the tip of chain B the new head
        let chain_b_tip_hash = chain_b.last().unwrap().hash();

        // verify the new canonical head
        test_harness.check_canon_head(chain_b_tip_hash);

        // verify the canonical head didn't change
        test_harness.check_canon_head(chain_b_tip_hash);
    }

    #[tokio::test]
    async fn test_engine_tree_reorg_with_missing_ancestor_expecting_valid() {
        reth_tracing::init_test_tracing();
        let chain_spec = MAINNET.clone();
        let mut test_harness = TestHarness::new(chain_spec.clone());

        let base_chain: Vec<_> = test_harness.block_builder.get_executed_blocks(0..6).collect();
        test_harness = test_harness.with_blocks(base_chain.clone());

        // create a side chain with an invalid block
        let side_chain = test_harness
            .block_builder
            .create_fork(base_chain.last().unwrap().recovered_block(), 15);
        let invalid_index = 9;

        test_harness.setup_range_insertion_for_invalid_chain(side_chain.clone(), invalid_index);

        for (index, block) in side_chain.iter().enumerate() {
            test_harness.send_new_payload(block.clone()).await;

            if index < side_chain.len() - invalid_index - 1 {
                test_harness.send_fcu(block.hash(), ForkchoiceStatus::Valid).await;
            }
        }

        // Try to do a forkchoice update to a block after the invalid one
        let fork_tip_hash = side_chain.last().unwrap().hash();
        test_harness.send_fcu(fork_tip_hash, ForkchoiceStatus::Invalid).await;
    }
}