use super::*;
use crate::{
    persistence::PersistenceAction,
    tree::{
        payload_validator::{BasicEngineValidator, TreeCtx, ValidationOutcome},
        persistence_state::CurrentPersistenceAction,
        PersistTarget, TreeConfig,
    },
};
use reth_storage_overlay::OverlayManager;

use alloy_eips::eip1898::BlockWithParent;
use alloy_primitives::{
    map::{B256Map, B256Set},
    Bytes, B256,
};
use alloy_rlp::Decodable;
use alloy_rpc_types_engine::{
    ExecutionData, ExecutionPayloadSidecar, ExecutionPayloadV1, ForkchoiceState,
    ForkchoiceUpdateError,
};
use assert_matches::assert_matches;
use reth_chain_state::{test_utils::TestBlockBuilder, BlockState};
use reth_chainspec::{ChainSpec, HOLESKY, MAINNET};
use reth_engine_primitives::{EngineApiValidator, ForkchoiceStatus, NoopInvalidBlockHook};
use reth_ethereum_consensus::EthBeaconConsensus;
use reth_ethereum_engine_primitives::{EthEngineTypes, EthPayloadAttributes};
use reth_ethereum_primitives::{Block, EthPrimitives};
use reth_evm_ethereum::MockEvmConfig;
use reth_payload_builder::PayloadServiceCommand;
use reth_primitives_traits::Block as _;
use reth_provider::{test_utils::MockEthProvider, BalStoreHandle, InMemoryBalStore, RawBal};
use reth_tasks::spawn_os_thread;
use reth_trie_common::ComputedTrieData;
use std::{
    collections::BTreeMap,
    str::FromStr,
    sync::{
        mpsc::{Receiver, Sender},
        Arc,
    },
    time::Duration,
};
use tokio::sync::oneshot;

/// Mock engine validator for tests
#[derive(Debug, Clone)]
struct MockEngineValidator;

impl reth_engine_primitives::PayloadValidator<EthEngineTypes> for MockEngineValidator {
    type Block = Block;

    fn convert_payload_to_block(
        &self,
        payload: ExecutionData,
    ) -> Result<
        reth_primitives_traits::SealedBlock<Self::Block>,
        reth_payload_primitives::NewPayloadError,
    > {
        let block = reth_ethereum_primitives::Block::try_from(payload.payload).map_err(|e| {
            reth_payload_primitives::NewPayloadError::Other(format!("{e:?}").into())
        })?;
        Ok(block.seal_slow())
    }
}

impl EngineApiValidator<EthEngineTypes> for MockEngineValidator {
    fn validate_version_specific_fields(
        &self,
        _version: reth_payload_primitives::EngineApiMessageVersion,
        _payload_or_attrs: reth_payload_primitives::PayloadOrAttributes<
            '_,
            alloy_rpc_types_engine::ExecutionData,
            alloy_rpc_types_engine::PayloadAttributes,
        >,
    ) -> Result<(), reth_payload_primitives::EngineObjectValidationError> {
        // Mock implementation - always valid
        Ok(())
    }

    fn ensure_well_formed_attributes(
        &self,
        _version: reth_payload_primitives::EngineApiMessageVersion,
        _attributes: &alloy_rpc_types_engine::PayloadAttributes,
    ) -> Result<(), reth_payload_primitives::EngineObjectValidationError> {
        // Mock implementation - always valid
        Ok(())
    }
}

/// 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.
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
    fn spawn_channel() -> (Sender<T>, Receiver<T>, TestChannelHandle) {
        use std::sync::mpsc::channel;
        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
    #[expect(dead_code)]
    fn release(&self) {
        let _ = self.release.send(());
    }
}

struct TestHarness {
    tree: EngineApiTreeHandler<
        EthPrimitives,
        MockEthProvider,
        EthEngineTypes,
        BasicEngineValidator<MockEthProvider, MockEvmConfig, MockEngineValidator>,
        MockEvmConfig,
    >,
    to_tree_tx: crossbeam_channel::Sender<
        FromEngine<EngineApiRequest<EthEngineTypes, EthPrimitives>, Block>,
    >,
    from_tree_rx: UnboundedReceiver<EngineApiEvent>,
    payload_command_rx: UnboundedReceiver<PayloadServiceCommand<EthEngineTypes>>,
    blocks: Vec<ExecutedBlock>,
    action_rx: Receiver<PersistenceAction>,
    block_builder: TestBlockBuilder,
    provider: MockEthProvider,
}

impl TestHarness {
    fn new(chain_spec: Arc<ChainSpec>) -> Self {
        Self::with_config(chain_spec, TreeConfig::default().with_has_enough_parallelism(true))
    }

    fn with_config(chain_spec: Arc<ChainSpec>, tree_config: TreeConfig) -> Self {
        use std::sync::mpsc::channel;
        let (action_tx, action_rx) = channel();
        Self::with_persistence_channel_and_config(chain_spec, action_tx, action_rx, tree_config)
    }

    #[expect(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 {
        Self::with_persistence_channel_and_config(
            chain_spec,
            action_tx,
            action_rx,
            TreeConfig::default().with_has_enough_parallelism(true),
        )
    }

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

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

        let provider = MockEthProvider::default();

        let payload_validator = MockEngineValidator;

        let (from_tree_tx, from_tree_rx) = unbounded_channel();
        let runtime = reth_tasks::Runtime::test();
        let overlay_manager = OverlayManager::new(runtime.state_trie_overlay_worker_pool());

        let header = chain_spec.genesis_header().clone();
        let header = SealedHeader::seal_slow(header);
        let engine_api_tree_state = EngineApiTreeState::new(
            10,
            10,
            tree_config.invalid_header_hit_eviction_threshold(),
            header.num_hash(),
            EngineApiKind::Ethereum,
            overlay_manager.clone(),
        );
        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 = MockEvmConfig::default();
        let engine_validator = BasicEngineValidator::new(
            provider.clone(),
            consensus.clone(),
            evm_config.clone(),
            payload_validator,
            tree_config.clone(),
            Box::new(NoopInvalidBlockHook::default()),
            overlay_manager,
            runtime.clone(),
        );

        let tree = EngineApiTreeHandler::new(
            provider.clone(),
            consensus,
            engine_validator,
            from_tree_tx,
            engine_api_tree_state,
            canonical_in_memory_state,
            persistence_handle,
            PersistenceState {
                last_persisted_block: BlockNumHash::default(),
                last_state_trie_persisted_block: BlockNumHash::default(),
                rx: None,
            },
            payload_builder,
            tree_config,
            EngineApiKind::Ethereum,
            evm_config,
            runtime,
        );

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

    fn with_blocks(mut self, blocks: Vec<ExecutedBlock>) -> Self {
        let mut blocks_by_hash = B256Map::default();
        let mut blocks_by_number = BTreeMap::new();
        let mut state_by_hash = B256Map::default();
        let mut hash_by_number = BTreeMap::new();
        let mut parent_to_child: B256Map<B256Set> = B256Map::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;
        }

        let overlay_manager = self.tree.state.tree_state.overlay_manager.clone();
        for block in &blocks {
            overlay_manager.insert_block(block.clone());
        }

        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,
            engine_kind: EngineApiKind::Ethereum,
            overlay_manager,
        };

        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
    }

    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();
        let _ = self
            .tree
            .on_engine_message(FromEngine::Request(
                BeaconEngineMessage::ForkchoiceUpdated {
                    state: fcu_state,
                    payload_attrs: None,
                    tx,
                }
                .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(ConsensusEngineEvent::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,
        }
    }

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

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

        self.provider.extend_blocks(block_data);
    }
}

/// Simplified test metrics for validation calls
#[derive(Debug, Default)]
struct TestMetrics {
    /// Count of successful `validate_block_direct` calls
    validation_calls: usize,
    /// Count of validation errors
    validation_errors: usize,
}

impl TestMetrics {
    fn record_validation(&mut self, success: bool) {
        if success {
            self.validation_calls += 1;
        } else {
            self.validation_errors += 1;
        }
    }

    fn total_calls(&self) -> usize {
        self.validation_calls + self.validation_errors
    }
}

/// Extended test harness with direct `validate_block_with_state` access
pub(crate) struct ValidatorTestHarness {
    /// Basic test harness
    harness: TestHarness,
    /// Direct access to validator for `validate_block_with_state` calls
    validator: BasicEngineValidator<MockEthProvider, MockEvmConfig, MockEngineValidator>,
    /// Simple validation metrics
    metrics: TestMetrics,
}

impl ValidatorTestHarness {
    fn new(chain_spec: Arc<ChainSpec>) -> Self {
        let harness = TestHarness::new(chain_spec.clone());

        // Create validator identical to the one in TestHarness
        let consensus = Arc::new(EthBeaconConsensus::new(chain_spec));
        let provider = harness.provider.clone();
        let payload_validator = MockEngineValidator;
        let evm_config = MockEvmConfig::default();
        let overlay_manager = harness.tree.state.tree_state.overlay_manager.clone();

        let validator = BasicEngineValidator::new(
            provider,
            consensus,
            evm_config,
            payload_validator,
            TreeConfig::default(),
            Box::new(NoopInvalidBlockHook::default()),
            overlay_manager,
            reth_tasks::Runtime::test(),
        );

        Self { harness, validator, metrics: TestMetrics::default() }
    }

    /// Configure `PersistenceState` for specific persistence scenarios
    fn start_persistence_operation(&mut self, action: CurrentPersistenceAction) {
        match action {
            CurrentPersistenceAction::SavingBlocks { highest } => {
                let (_tx, rx) = crossbeam_channel::bounded(1);
                self.harness.tree.persistence_state.start_save(highest, rx);
            }
            CurrentPersistenceAction::RemovingBlocks { new_tip_num } => {
                let (_tx, rx) = crossbeam_channel::bounded(1);
                self.harness.tree.persistence_state.start_remove(new_tip_num, rx);
            }
        }
    }

    /// Check if persistence is currently in progress
    fn is_persistence_in_progress(&self) -> bool {
        self.harness.tree.persistence_state.in_progress()
    }

    /// Call `validate_block_with_state` directly with block
    fn validate_block_direct(
        &mut self,
        block: SealedBlock<Block>,
    ) -> ValidationOutcome<EthPrimitives> {
        let ctx = TreeCtx::new(
            &mut self.harness.tree.state,
            &self.harness.tree.canonical_in_memory_state,
        );
        let result = self.validator.validate_block(block, ctx);
        self.metrics.record_validation(result.is_ok());
        result
    }

    /// Get validation metrics for testing
    fn validation_call_count(&self) -> usize {
        self.metrics.total_calls()
    }
}

/// Factory for creating test blocks with controllable properties
struct TestBlockFactory {
    builder: TestBlockBuilder,
}

impl TestBlockFactory {
    fn new(chain_spec: ChainSpec) -> Self {
        Self { builder: TestBlockBuilder::eth().with_chain_spec(chain_spec) }
    }

    /// Create block that triggers consensus violation by corrupting state root
    fn create_invalid_consensus_block(&mut self, parent_hash: B256) -> SealedBlock<Block> {
        let mut block = self.builder.generate_random_block(1, parent_hash).into_block();

        // Corrupt state root to trigger consensus violation
        block.header.state_root = B256::random();

        block.seal_slow()
    }

    /// Create block that triggers execution failure
    fn create_invalid_execution_block(&mut self, parent_hash: B256) -> SealedBlock<Block> {
        let mut block = self.builder.generate_random_block(1, parent_hash).into_block();

        // Create transaction that will fail execution
        // This is simplified - in practice we'd create a transaction with insufficient gas, etc.
        block.header.gas_used = block.header.gas_limit + 1; // Gas used exceeds limit

        block.seal_slow()
    }

    /// Create valid block
    fn create_valid_block(&mut self, parent_hash: B256) -> SealedBlock<Block> {
        let block = self.builder.generate_random_block(1, parent_hash).into_block();
        block.seal_slow()
    }
}

#[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 = match test_harness.tree.wait_for_event() {
        super::LoopEvent::EngineMessage(msg) => msg,
        other => panic!("unexpected event: {other:?}"),
    };
    let _ = test_harness.tree.on_engine_message(msg).unwrap();

    // we now should receive the other batch
    let msg = match test_harness.tree.wait_for_event() {
        super::LoopEvent::EngineMessage(msg) => msg,
        other => panic!("unexpected event: {other:?}"),
    };
    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());
    spawn_os_thread("engine", || test_harness.tree.run());

    // 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(input, _) = 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!(input.persist_rest_blocks().len(), expected_persist_len);
        assert_eq!(input.persist_rest_blocks(), &blocks[..expected_persist_len]);
        assert_eq!(input.prev_db_tip(), input.prev_partial_state_trie());
        assert_eq!(input.new_db_tip(), input.new_partial_state_trie());
    } else {
        panic!("unexpected action received {received_action:?}");
    }
}

#[test]
fn on_new_persisted_block_queues_sparse_trie_prune_request() {
    let blocks: Vec<_> = TestBlockBuilder::eth().get_executed_blocks(1..4).collect();
    let mut test_harness = TestHarness::new(MAINNET.clone()).with_blocks(blocks.clone());
    let persisted = blocks[0].recovered_block().num_hash();
    test_harness.tree.persistence_state.finish(persisted, persisted);

    test_harness.tree.on_new_persisted_block().unwrap();

    assert!(test_harness.tree.state.pending_sparse_trie_prune());
}

#[test]
fn on_new_persisted_block_queues_sparse_trie_prune_with_in_memory_blocks() {
    let blocks: Vec<_> = TestBlockBuilder::eth().get_executed_blocks(1..4).collect();
    let mut test_harness = TestHarness::new(MAINNET.clone()).with_blocks(blocks.clone());
    let persisted = blocks[0].recovered_block().num_hash();
    test_harness.tree.persistence_state.finish(persisted, persisted);

    test_harness.tree.on_new_persisted_block().unwrap();

    assert!(test_harness.tree.state.pending_sparse_trie_prune());
}

#[test]
fn on_new_persisted_block_skips_sparse_trie_prune_when_state_root_task_disabled() {
    let blocks: Vec<_> = TestBlockBuilder::eth().get_executed_blocks(1..4).collect();
    let configs = [
        TreeConfig::default().with_has_enough_parallelism(false),
        TreeConfig::default().with_has_enough_parallelism(true).with_state_root_fallback(true),
        TreeConfig::default().with_has_enough_parallelism(true).with_skip_state_root(true),
    ];

    for config in configs {
        let mut test_harness = TestHarness::new(MAINNET.clone()).with_blocks(blocks.clone());
        test_harness.tree.config = config;
        let persisted = blocks[0].recovered_block().num_hash();
        test_harness.tree.persistence_state.finish(persisted, persisted);

        test_harness.tree.on_new_persisted_block().unwrap();

        assert!(!test_harness.tree.state.pending_sparse_trie_prune());
    }
}

#[test]
fn payload_build_tracker_notifies_after_last_lease_drops() {
    let (tracker, finished) = PayloadBuildTracker::new();
    let first = tracker.acquire();
    let second = tracker.acquire();

    assert!(tracker.is_active());
    drop(first);
    assert!(tracker.is_active());
    assert!(finished.try_recv().is_err());

    drop(second);
    assert!(!tracker.is_active());
    assert_eq!(finished.try_recv(), Ok(()));
}

#[test]
fn persistence_handoff_waits_for_active_payload_jobs() {
    let config = TreeConfig::default()
        .with_has_enough_parallelism(true)
        .with_persistence_threshold(0)
        .with_memory_block_buffer_target(0);
    let blocks: Vec<_> = TestBlockBuilder::eth().get_executed_blocks(1..4).collect();
    let mut test_harness =
        TestHarness::with_config(MAINNET.clone(), config).with_blocks(blocks.clone());
    let persisted = blocks[0].recovered_block().num_hash();
    let persisted_hash = persisted.hash;
    let first = test_harness.tree.payload_builds.acquire();
    let second = test_harness.tree.payload_builds.acquire();

    test_harness
        .tree
        .on_persistence_complete(
            PersistenceResult {
                last_block: persisted,
                last_state_trie_block: persisted,
                commit_duration: Some(Duration::ZERO),
            },
            Instant::now(),
        )
        .unwrap();

    // Durability advances immediately, but the in-memory overlay remains available to jobs.
    assert_eq!(test_harness.tree.persistence_state.last_persisted_block, persisted);
    assert!(test_harness.tree.pending_persisted_handoff.is_some());
    assert!(test_harness.tree.state.tree_state.contains_hash(&persisted_hash));
    assert!(test_harness.tree.canonical_in_memory_state.state_by_hash(persisted_hash).is_some());

    // Do not let another persistence action overwrite the deferred handoff's frontier.
    test_harness.tree.advance_persistence().unwrap();
    assert!(!test_harness.tree.persistence_state.in_progress());

    drop(first);
    assert!(test_harness.tree.payload_build_finished.try_recv().is_err());
    assert!(test_harness.tree.pending_persisted_handoff.is_some());

    drop(second);
    assert!(matches!(test_harness.tree.wait_for_event(), super::LoopEvent::PayloadBuildFinished));
    test_harness.tree.on_payload_build_finished().unwrap();

    assert!(test_harness.tree.pending_persisted_handoff.is_none());
    assert!(!test_harness.tree.state.tree_state.contains_hash(&persisted_hash));
    assert!(test_harness.tree.canonical_in_memory_state.state_by_hash(persisted_hash).is_none());
}

#[test]
fn pending_persisted_handoff_blocks_engine_messages_until_payload_build_finishes() {
    let blocks: Vec<_> = TestBlockBuilder::eth().get_executed_blocks(1..4).collect();
    let mut test_harness = TestHarness::with_config(
        MAINNET.clone(),
        TreeConfig::default().with_has_enough_parallelism(true),
    )
    .with_blocks(blocks.clone());
    let persisted = blocks[0].recovered_block().num_hash();
    let payload_build = test_harness.tree.payload_builds.acquire();

    test_harness
        .tree
        .on_persistence_complete(
            PersistenceResult {
                last_block: persisted,
                last_state_trie_block: persisted,
                commit_duration: Some(Duration::ZERO),
            },
            Instant::now(),
        )
        .unwrap();
    assert!(test_harness.tree.pending_persisted_handoff.is_some());

    // Forkchoice updates received while the handoff is pending could start replacement payload
    // builds and keep the final completion notification from reaching the tree. Queue a generic
    // engine message to prove the gate leaves every message upstream until then.
    test_harness.to_tree_tx.send(FromEngine::Event(FromOrchestrator::BackfillSyncStarted)).unwrap();

    let (started_tx, started_rx) = std::sync::mpsc::sync_channel(0);
    let (event_tx, event_rx) = std::sync::mpsc::sync_channel(1);
    std::thread::scope(|scope| {
        let tree = &mut test_harness.tree;
        scope.spawn(move || {
            started_tx.send(()).unwrap();
            event_tx
                .send(matches!(tree.wait_for_event(), super::LoopEvent::PayloadBuildFinished))
                .unwrap();
        });

        started_rx.recv().unwrap();
        assert!(
            event_rx.recv_timeout(Duration::from_millis(100)).is_err(),
            "a pending handoff must prevent queued engine messages from being processed"
        );

        drop(payload_build);
        assert_eq!(event_rx.recv_timeout(Duration::from_secs(1)), Ok(true));
    });

    test_harness.tree.on_payload_build_finished().unwrap();
    assert!(test_harness.tree.pending_persisted_handoff.is_none());
    assert_matches!(
        test_harness.tree.wait_for_event(),
        super::LoopEvent::EngineMessage(FromEngine::Event(FromOrchestrator::BackfillSyncStarted))
    );
}

#[test]
fn persist_until_complete_updates_frontiers_without_in_memory_handoff() {
    let blocks: Vec<_> = TestBlockBuilder::eth().get_executed_blocks(1..4).collect();
    let mut test_harness = TestHarness::new(MAINNET.clone()).with_blocks(blocks.clone());
    let persisted = blocks.last().unwrap().recovered_block().num_hash();
    let persisted_hash = persisted.hash;
    let (tx, rx) = crossbeam_channel::bounded(1);

    tx.send(PersistenceResult {
        last_block: persisted,
        last_state_trie_block: persisted,
        commit_duration: Some(Duration::ZERO),
    })
    .unwrap();
    test_harness.tree.persistence_state.start_save(persisted, rx);

    test_harness.tree.persist_until_complete().unwrap();

    // Shutdown waits for the durable write, but process teardown does not need the destructive
    // in-memory handoff that normally follows a completed persistence task.
    assert_eq!(test_harness.tree.persistence_state.last_persisted_block, persisted);
    assert_eq!(test_harness.tree.persistence_state.last_state_trie_persisted_block, persisted);
    assert!(!test_harness.tree.persistence_state.in_progress());
    assert!(test_harness.tree.pending_persisted_handoff.is_none());
    assert!(test_harness.tree.state.tree_state.contains_hash(&persisted_hash));
    assert!(test_harness.tree.canonical_in_memory_state.state_by_hash(persisted_hash).is_some());
}

#[test]
fn backfill_action_skips_while_payload_build_is_active() {
    let mut test_harness = TestHarness::new(MAINNET.clone());
    let payload_build = test_harness.tree.payload_builds.acquire();
    let action = BackfillAction::Start(B256::random().into());

    test_harness.tree.emit_event(EngineApiEvent::BackfillAction(action));
    assert!(test_harness.tree.pending_persisted_handoff.is_none());
    assert!(test_harness.tree.backfill_sync_state.is_idle());
    assert!(test_harness.from_tree_rx.try_recv().is_err());

    drop(payload_build);
    assert!(matches!(test_harness.tree.wait_for_event(), super::LoopEvent::PayloadBuildFinished));
    test_harness.tree.on_payload_build_finished().unwrap();

    // The skipped action is not queued for a later replay.
    assert!(test_harness.tree.backfill_sync_state.is_idle());
    assert!(test_harness.from_tree_rx.try_recv().is_err());
}

#[test]
fn backfill_action_skips_while_persisted_handoff_is_pending() {
    let blocks: Vec<_> = TestBlockBuilder::eth().get_executed_blocks(1..4).collect();
    let mut test_harness = TestHarness::with_config(
        MAINNET.clone(),
        TreeConfig::default().with_has_enough_parallelism(true),
    )
    .with_blocks(blocks.clone());
    let persisted = blocks[0].recovered_block().num_hash();
    let payload_build = test_harness.tree.payload_builds.acquire();

    test_harness
        .tree
        .on_persistence_complete(
            PersistenceResult {
                last_block: persisted,
                last_state_trie_block: persisted,
                commit_duration: Some(Duration::ZERO),
            },
            Instant::now(),
        )
        .unwrap();

    drop(payload_build);
    assert!(!test_harness.tree.payload_builds.is_active());
    assert!(test_harness.tree.pending_persisted_handoff.is_some());

    test_harness
        .tree
        .emit_event(EngineApiEvent::BackfillAction(BackfillAction::Start(B256::random().into())));

    assert!(test_harness.tree.backfill_sync_state.is_idle());
    assert!(test_harness.from_tree_rx.try_recv().is_err());
}

#[test]
fn configured_persistence_suppression_tracks_payload_job_lifetime() {
    let config = TreeConfig::default()
        .with_has_enough_parallelism(true)
        .with_persistence_threshold(0)
        .with_memory_block_buffer_target(0)
        .with_suppress_persistence_during_build(true);
    let blocks: Vec<_> = TestBlockBuilder::eth().get_executed_blocks(1..4).collect();
    let test_harness = TestHarness::with_config(MAINNET.clone(), config).with_blocks(blocks);
    let payload_build = test_harness.tree.payload_builds.acquire();

    assert!(test_harness.tree.get_save_blocks_input(PersistTarget::Threshold).is_none());

    drop(payload_build);
    assert!(test_harness.tree.get_save_blocks_input(PersistTarget::Threshold).is_some());
}

#[test]
fn remove_blocks_clears_pending_sparse_trie_prune_request() {
    let mut test_harness = TestHarness::new(MAINNET.clone());
    test_harness.tree.persistence_state.last_persisted_block =
        BlockNumHash { hash: B256::random(), number: 10 };
    test_harness.tree.state.set_pending_sparse_trie_prune(true);

    test_harness.tree.remove_blocks(9);

    assert!(!test_harness.tree.state.pending_sparse_trie_prune());
}

#[test]
fn process_payload_attributes_shares_sparse_trie_during_validation_fallback() {
    let config = TreeConfig::default()
        .with_has_enough_parallelism(true)
        .with_state_root_fallback(true)
        .with_share_sparse_trie_with_payload_builder(true);
    let blocks: Vec<_> = TestBlockBuilder::eth().get_executed_blocks(1..2).collect();
    let mut test_harness = TestHarness::with_config(MAINNET.clone(), config).with_blocks(blocks);
    let head =
        test_harness.blocks.last().unwrap().recovered_block().clone_sealed_header().clone_header();
    let head_hash = test_harness.blocks.last().unwrap().recovered_block().hash();
    let state = test_harness.fcu_state(head_hash);
    test_harness.tree.state.set_pending_sparse_trie_prune(true);

    let updated = test_harness.tree.process_payload_attributes(
        EthPayloadAttributes {
            timestamp: head.timestamp() + 1,
            prev_randao: B256::ZERO,
            suggested_fee_recipient: Default::default(),
            withdrawals: None,
            parent_beacon_block_root: None,
            slot_number: None,
            target_gas_limit: None,
        },
        &head,
        state,
    );

    assert_eq!(updated.forkchoice_status(), ForkchoiceStatus::Valid);
    assert!(!test_harness.tree.state.pending_sparse_trie_prune());

    let command = test_harness.payload_command_rx.try_recv().unwrap();
    let PayloadServiceCommand::BuildNewPayload(input, _, _) = command else {
        panic!("expected build new payload command")
    };
    assert!(input.resources.state_root_handle().is_some());
    assert!(test_harness.tree.payload_builds.is_active());

    drop(input);
    assert!(!test_harness.tree.payload_builds.is_active());
}

#[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();
    let _ = 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,
            }
            .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 sealed_clone = sealed.clone();
    let block = sealed.into_block();
    let payload = ExecutionPayloadV1::from_block_unchecked(hash, &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, sealed_clone);
}

#[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();

    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()
        })
    );
}

#[test]
fn test_validated_payload_bal_is_inserted_into_store() {
    let mut block_builder = TestBlockBuilder::eth();
    let parent = block_builder.get_executed_block_with_number(1, B256::ZERO);
    let child = block_builder.get_executed_block_with_number(2, parent.recovered_block().hash());
    let child_block = child.recovered_block().clone_sealed_block();
    let child_num_hash = child_block.num_hash();
    let raw_bal = Bytes::from_static(&[0xc0]);

    let mut test_harness = TestHarness::new(MAINNET.clone()).with_blocks(vec![parent]);
    let bal_store = BalStoreHandle::new(InMemoryBalStore::default());
    test_harness.tree.provider.bal_store = bal_store.clone();

    let outcome = test_harness
        .tree
        .insert_block_or_payload(
            child_block.block_with_parent(),
            child,
            |_, executed, _| {
                Ok::<_, InsertPayloadError<Block>>(
                    ValidationOutput::new(executed, None)
                        .with_raw_bal(Some(RawBal::from(raw_bal.clone()))),
                )
            },
            |_, executed| Ok(executed.recovered_block().clone_sealed_block()),
        )
        .unwrap();

    assert_eq!(outcome, InsertPayloadOk::Inserted(BlockStatus::Valid));
    assert_eq!(bal_store.get_by_hash(child_num_hash.hash).unwrap(), Some(raw_bal));
}

#[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 hash = sealed.hash();
    let block = sealed.into_block();
    let payload = ExecutionPayloadV1::from_block_unchecked(hash, &block);

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

    let (tx, rx) = oneshot::channel();
    let _ = 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_backpressure_waits_for_persistence_before_reading_incoming() {
    let blocks: Vec<_> = TestBlockBuilder::eth().get_executed_blocks(1..4).collect();
    let mut test_harness = TestHarness::new(MAINNET.clone()).with_blocks(blocks.clone());
    test_harness.tree.config = test_harness
        .tree
        .config
        .with_persistence_threshold(0)
        .with_memory_block_buffer_target(0)
        .with_persistence_backpressure_threshold(1);

    let (persist_tx, persist_rx) = crossbeam_channel::bounded(1);
    let persisted = blocks.last().unwrap().recovered_block().num_hash();
    test_harness.tree.persistence_state.start_save(persisted, persist_rx);
    assert!(test_harness.tree.should_backpressure());

    let (tx, mut rx) = oneshot::channel();
    test_harness
        .to_tree_tx
        .send(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,
            }
            .into(),
        ))
        .unwrap();
    test_harness.to_tree_tx.send(FromEngine::DownloadedBlocks(vec![])).unwrap();
    assert_eq!(test_harness.tree.incoming.len(), 2);

    std::thread::spawn(move || {
        std::thread::sleep(Duration::from_millis(10));
        persist_tx
            .send(PersistenceResult {
                last_block: persisted,
                last_state_trie_block: persisted,
                commit_duration: Some(Duration::ZERO),
            })
            .unwrap();
    });

    let event = test_harness.tree.wait_for_persistence_event();
    assert!(matches!(event, super::LoopEvent::PersistenceComplete { .. }));
    assert_eq!(test_harness.tree.incoming.len(), 2);

    let super::LoopEvent::PersistenceComplete { result, start_time } = event else {
        unreachable!()
    };
    test_harness.tree.on_persistence_complete(result, start_time).unwrap();

    let super::LoopEvent::EngineMessage(message) = test_harness.tree.wait_for_event() else {
        panic!("expected queued engine message")
    };
    let _ = test_harness.tree.on_engine_message(message).unwrap();
    let msg = rx.try_recv();
    assert!(msg.is_ok());
    assert_eq!(test_harness.tree.incoming.len(), 1);

    let super::LoopEvent::EngineMessage(message) = test_harness.tree.wait_for_event() else {
        panic!("expected queued engine message")
    };
    let _ = test_harness.tree.on_engine_message(message).unwrap();
    assert_eq!(test_harness.tree.incoming.len(), 0);
}

#[test]
fn test_backpressure_excludes_in_memory_buffer() {
    for (canonical_tip, expected_backpressure) in [(14_u64, false), (15, true)] {
        let blocks: Vec<_> =
            TestBlockBuilder::eth().get_executed_blocks(1..canonical_tip + 1).collect();
        let mut test_harness = TestHarness::new(MAINNET.clone()).with_blocks(blocks.clone());
        test_harness.tree.config = test_harness
            .tree
            .config
            .with_persistence_threshold(0)
            .with_memory_block_buffer_target(5)
            .with_persistence_backpressure_threshold(10);

        let (_persist_tx, persist_rx) = crossbeam_channel::bounded(1);
        let persisted = blocks.last().unwrap().recovered_block().num_hash();
        test_harness.tree.persistence_state.start_save(persisted, persist_rx);

        assert_eq!(test_harness.tree.should_backpressure(), expected_backpressure);
    }
}

#[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(input, sender) = received_action else {
        panic!("received wrong action");
    };
    assert_eq!(input.persist_rest_blocks(), &blocks[..2]);

    // send the response so we can advance again
    sender
        .send(PersistenceResult {
            last_block: blocks[1].recovered_block().num_hash(),
            last_state_trie_block: blocks[1].recovered_block().num_hash(),
            commit_duration: Some(Duration::ZERO),
        })
        .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 polling persistence completion, we should be at `None` for the next action
    test_harness.tree.try_poll_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);

    let empty_trie_data = ComputedTrieData::default;

    for block in &chain_a {
        test_harness.tree.state.tree_state.insert_executed(ExecutedBlock::new(
            Arc::new(block.clone()),
            Arc::new(BlockExecutionOutput::default()),
            empty_trie_data(),
        ));
    }
    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(ExecutedBlock::new(
            Arc::new(block.clone()),
            Arc::new(BlockExecutionOutput::default()),
            empty_trie_data(),
        ));
    }

    // 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 =
        blocks[last_persisted_block_number as usize].recovered_block.num_hash();
    test_harness.tree.persistence_state.last_state_trie_persisted_block =
        blocks[last_persisted_block_number as usize].recovered_block.num_hash();

    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 input = test_harness.tree.get_save_blocks_input(PersistTarget::Threshold).unwrap();
    let blocks_to_persist = input.persist_rest_blocks();

    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.sealed_header_by_hash(&fork_block_hash).is_some());

    let input = test_harness.tree.get_save_blocks_input(PersistTarget::Threshold).unwrap();
    let blocks_to_persist = input.persist_rest_blocks();
    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()
        })
    );
}

#[test]
fn test_threshold_persistence_with_state_masking_blocks() {
    let mut test_harness = TestHarness::new(MAINNET.clone());
    let blocks: Vec<_> = TestBlockBuilder::eth().get_executed_blocks(0..9).collect();
    test_harness = test_harness.with_blocks(blocks.clone());
    test_harness.tree.persistence_state.last_state_trie_persisted_block =
        blocks[1].recovered_block().num_hash();
    test_harness.tree.persistence_state.last_persisted_block =
        blocks[3].recovered_block().num_hash();
    test_harness.tree.config = TreeConfig::default()
        .with_persistence_threshold(4)
        .with_memory_block_buffer_target(1)
        .with_num_state_masking_blocks(2);

    test_harness.tree.advance_persistence().unwrap();
    let action = test_harness.action_rx.recv().unwrap();
    let PersistenceAction::SaveBlocks(input, sender) = action else {
        panic!("expected save blocks action, got {action:?}")
    };

    let persisted_tip = blocks[7].recovered_block().num_hash();
    let state_trie_tip = blocks[5].recovered_block().num_hash();
    assert_eq!(input.new_db_tip(), persisted_tip.number);
    assert_eq!(input.new_partial_state_trie(), state_trie_tip.number);
    sender
        .send(PersistenceResult {
            last_block: persisted_tip,
            last_state_trie_block: state_trie_tip,
            commit_duration: Some(Duration::ZERO),
        })
        .unwrap();
    test_harness.tree.try_poll_persistence().unwrap();

    assert_eq!(test_harness.tree.persistence_state.last_persisted_block, persisted_tip);
    assert_eq!(test_harness.tree.persistence_state.last_state_trie_persisted_block, state_trie_tip);
    assert_eq!(
        test_harness.tree.canonical_in_memory_state.get_persisted_num_hash(),
        Some(persisted_tip)
    );

    let removed_hash = blocks[state_trie_tip.number as usize].recovered_block().hash();
    let retained_hash = blocks[(state_trie_tip.number + 1) as usize].recovered_block().hash();
    assert!(test_harness.tree.state.tree_state.executed_block_by_hash(removed_hash).is_none());
    assert!(test_harness.tree.state.tree_state.executed_block_by_hash(retained_hash).is_some());
    assert!(test_harness.tree.canonical_in_memory_state.state_by_hash(removed_hash).is_none());
    assert!(test_harness.tree.canonical_in_memory_state.state_by_hash(retained_hash).is_some());
}

#[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 = B256Set::from_iter([missing_block.hash()]);
            assert_eq!(actual_block_set, expected_block_set);
        }
        _ => panic!("Unexpected event: {event:#?}"),
    }
}

#[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());
    let _ = 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, B256Set::from_iter([main_chain_last_hash]));
        }
        _ => panic!("Unexpected event: {event:#?}"),
    }

    // After backfill completes with head not buffered, we also request head download
    let event = test_harness.from_tree_rx.recv().await.unwrap();
    match event {
        EngineApiEvent::Download(DownloadRequest::BlockSet(hash_set)) => {
            assert_eq!(hash_set, B256Set::from_iter([main_chain_last_hash]));
        }
        _ => panic!("Unexpected event: {event:#?}"),
    }

    let _ = test_harness
        .tree
        .on_engine_message(FromEngine::DownloadedBlocks(vec![main_chain
            .last()
            .unwrap()
            .clone_sealed_block()]))
        .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_fcu_with_canonical_ancestor_updates_latest_block() {
    // Test for issue where FCU with canonical ancestor doesn't update Latest block state
    // This was causing "nonce too low" errors when discard_reorged_transactions is enabled

    reth_tracing::init_test_tracing();
    let chain_spec = MAINNET.clone();

    // Create test harness
    let mut test_harness = TestHarness::new(chain_spec.clone());

    // Set engine kind to OpStack and enable unwind_canonical_header to ensure the fix is triggered
    test_harness.tree.engine_kind = EngineApiKind::OpStack;
    test_harness.tree.config = test_harness.tree.config.clone().with_unwind_canonical_header(true);
    let mut test_block_builder = TestBlockBuilder::eth().with_chain_spec((*chain_spec).clone());

    // Create a chain of blocks
    let blocks: Vec<_> = test_block_builder.get_executed_blocks(1..5).collect();
    test_harness = test_harness.with_blocks(blocks.clone());

    // Set block 4 as the current canonical head
    let current_head = blocks[3].recovered_block().clone(); // Block 4 (0-indexed as blocks[3])
    let current_head_sealed = current_head.clone_sealed_header();
    test_harness.tree.state.tree_state.set_canonical_head(current_head.num_hash());
    test_harness.tree.canonical_in_memory_state.set_canonical_head(current_head_sealed);

    // Verify the current head is set correctly
    assert_eq!(test_harness.tree.state.tree_state.canonical_block_number(), current_head.number());
    assert_eq!(test_harness.tree.state.tree_state.canonical_block_hash(), current_head.hash());

    // Now perform FCU to a canonical ancestor (block 2)
    let ancestor_block = blocks[1].recovered_block().clone(); // Block 2 (0-indexed as blocks[1])

    // Send FCU to the canonical ancestor
    let (tx, rx) = oneshot::channel();
    let _ = test_harness
        .tree
        .on_engine_message(FromEngine::Request(
            BeaconEngineMessage::ForkchoiceUpdated {
                state: ForkchoiceState {
                    head_block_hash: ancestor_block.hash(),
                    safe_block_hash: B256::ZERO,
                    finalized_block_hash: B256::ZERO,
                },
                payload_attrs: None,
                tx,
            }
            .into(),
        ))
        .unwrap();

    // Verify FCU succeeds
    let response = rx.await.unwrap().unwrap().await.unwrap();
    assert!(response.payload_status.is_valid());

    // The critical test: verify that Latest block has been updated to the canonical ancestor
    // Check tree state
    assert_eq!(
        test_harness.tree.state.tree_state.canonical_block_number(),
        ancestor_block.number(),
        "Tree state: Latest block number should be updated to canonical ancestor"
    );
    assert_eq!(
        test_harness.tree.state.tree_state.canonical_block_hash(),
        ancestor_block.hash(),
        "Tree state: Latest block hash should be updated to canonical ancestor"
    );

    // Also verify canonical in-memory state is synchronized
    assert_eq!(
        test_harness.tree.canonical_in_memory_state.get_canonical_head().number,
        ancestor_block.number(),
        "In-memory state: Latest block number should be updated to canonical ancestor"
    );
    assert_eq!(
        test_harness.tree.canonical_in_memory_state.get_canonical_head().hash(),
        ancestor_block.hash(),
        "In-memory state: Latest block hash should be updated to canonical ancestor"
    );
}

#[tokio::test]
async fn test_fcu_with_canonical_ancestor_below_finalized_is_rejected() {
    reth_tracing::init_test_tracing();
    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(1..5).collect();
    test_harness = test_harness.with_blocks(blocks.clone());

    let current_head = blocks[3].recovered_block();
    let finalized = blocks[2].recovered_block();
    let ancestor = blocks[1].recovered_block();
    test_harness.tree.canonical_in_memory_state.set_finalized(finalized.clone_sealed_header());

    // An FCU to an ancestor below the latest known finalized block would reorg out the
    // finalized block and is rejected, with or without payload attributes.
    let payload_attributes = EthPayloadAttributes {
        timestamp: ancestor.timestamp() + 1,
        prev_randao: B256::ZERO,
        suggested_fee_recipient: Default::default(),
        withdrawals: None,
        parent_beacon_block_root: None,
        slot_number: None,
        target_gas_limit: None,
    };
    for attrs in [Some(payload_attributes), None] {
        let err = test_harness
            .tree
            .on_forkchoice_updated(
                ForkchoiceState {
                    head_block_hash: ancestor.hash(),
                    safe_block_hash: B256::ZERO,
                    finalized_block_hash: B256::ZERO,
                },
                attrs,
            )
            .unwrap()
            .outcome
            .await
            .unwrap_err();

        assert_matches!(err, ForkchoiceUpdateError::TooDeepReorg);
    }

    // no payload build is started and the canonical head remains untouched
    assert!(test_harness.payload_command_rx.try_recv().is_err());
    assert_eq!(test_harness.tree.state.tree_state.canonical_block_hash(), current_head.hash());

    // the finalized block itself is not below finality and can become the parent of the next
    // block
    let payload_attributes = EthPayloadAttributes {
        timestamp: finalized.timestamp() + 1,
        prev_randao: B256::ZERO,
        suggested_fee_recipient: Default::default(),
        withdrawals: None,
        parent_beacon_block_root: None,
        slot_number: None,
        target_gas_limit: None,
    };
    let outcome = test_harness
        .tree
        .on_forkchoice_updated(
            ForkchoiceState {
                head_block_hash: finalized.hash(),
                safe_block_hash: B256::ZERO,
                finalized_block_hash: B256::ZERO,
            },
            Some(payload_attributes),
        )
        .unwrap();

    assert_eq!(outcome.outcome.forkchoice_status(), ForkchoiceStatus::Valid);
    let command = test_harness.payload_command_rx.try_recv().unwrap();
    let PayloadServiceCommand::BuildNewPayload(input, _, _) = command else {
        panic!("expected build new payload command")
    };
    assert_eq!(input.parent_hash, finalized.hash());
}

#[tokio::test]
async fn test_fcu_with_canonical_ancestor_above_finalized_starts_payload_build() {
    reth_tracing::init_test_tracing();
    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(1..5).collect();
    test_harness = test_harness.with_blocks(blocks.clone());

    let current_head = blocks[3].recovered_block();
    let finalized = blocks[0].recovered_block();
    let ancestor = blocks[2].recovered_block();
    test_harness.tree.canonical_in_memory_state.set_finalized(finalized.clone_sealed_header());

    let payload_attributes = EthPayloadAttributes {
        timestamp: ancestor.timestamp() + 1,
        prev_randao: B256::ZERO,
        suggested_fee_recipient: Default::default(),
        withdrawals: None,
        parent_beacon_block_root: None,
        slot_number: None,
        target_gas_limit: None,
    };
    let outcome = test_harness
        .tree
        .on_forkchoice_updated(
            ForkchoiceState {
                head_block_hash: ancestor.hash(),
                safe_block_hash: B256::ZERO,
                finalized_block_hash: B256::ZERO,
            },
            Some(payload_attributes),
        )
        .unwrap();

    assert_eq!(outcome.outcome.forkchoice_status(), ForkchoiceStatus::Valid);

    // a payload build is started on top of the ancestor
    let command = test_harness.payload_command_rx.try_recv().unwrap();
    let PayloadServiceCommand::BuildNewPayload(input, _, _) = command else {
        panic!("expected build new payload command")
    };
    assert_eq!(input.parent_hash, ancestor.hash());

    // the canonical chain remains untouched, only the built block reorgs it eventually
    assert_eq!(test_harness.tree.state.tree_state.canonical_block_hash(), current_head.hash());
    assert_eq!(
        test_harness.tree.canonical_in_memory_state.get_canonical_head().hash(),
        current_head.hash()
    );
}

/// Test that verifies the happy path where a new payload extends the canonical chain
#[test]
fn test_on_new_payload_canonical_insertion() {
    reth_tracing::init_test_tracing();

    // Use test data similar to test_disconnected_payload
    let s = include_str!("../../test-data/holesky/1.rlp");
    let data = Bytes::from_str(s).unwrap();
    let block1 = Block::decode(&mut data.as_ref()).unwrap();
    let sealed1 = block1.seal_slow();
    let hash1 = sealed1.hash();
    let sealed1_clone = sealed1.clone();
    let block1 = sealed1.into_block();
    let payload1 = ExecutionPayloadV1::from_block_unchecked(hash1, &block1);

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

    // Case 1: Submit payload when NOT sync target head - should be syncing (disconnected)
    let outcome1 = test_harness
        .tree
        .on_new_payload(ExecutionData {
            payload: payload1.into(),
            sidecar: ExecutionPayloadSidecar::none(),
        })
        .unwrap();

    // Since this is disconnected from genesis, it should be syncing
    assert!(outcome1.outcome.is_syncing(), "Disconnected payload should be syncing");

    // Verify no canonicalization event
    assert!(outcome1.event.is_none(), "Should not trigger canonicalization when syncing");

    // Ensure block is buffered (like test_disconnected_payload)
    let buffered = test_harness.tree.state.buffer.block(&hash1).unwrap();
    assert_eq!(buffered.clone(), sealed1_clone, "Block should be buffered");
}

/// Test that ensures payloads are rejected when linking to a known-invalid ancestor
#[test]
fn test_on_new_payload_invalid_ancestor() {
    reth_tracing::init_test_tracing();

    // Use Holesky test data
    let mut test_harness = TestHarness::new(HOLESKY.clone());

    // Read block 1 from test data
    let s1 = include_str!("../../test-data/holesky/1.rlp");
    let data1 = Bytes::from_str(s1).unwrap();
    let block1 = Block::decode(&mut data1.as_ref()).unwrap();
    let sealed1 = block1.seal_slow();
    let hash1 = sealed1.hash();
    let parent1 = sealed1.parent_hash();

    // Mark block 1 as invalid
    test_harness
        .tree
        .state
        .invalid_headers
        .insert(BlockWithParent { block: sealed1.num_hash(), parent: parent1 });

    // Read block 2 which has block 1 as parent
    let s2 = include_str!("../../test-data/holesky/2.rlp");
    let data2 = Bytes::from_str(s2).unwrap();
    let block2 = Block::decode(&mut data2.as_ref()).unwrap();
    let sealed2 = block2.seal_slow();
    let hash2 = sealed2.hash();

    // Verify block2's parent is block1
    assert_eq!(sealed2.parent_hash(), hash1, "Block 2 should have block 1 as parent");

    let payload2 = ExecutionPayloadV1::from_block_unchecked(hash2, &sealed2.into_block());

    // Submit payload 2 (child of invalid block 1)
    let outcome = test_harness
        .tree
        .on_new_payload(ExecutionData {
            payload: payload2.into(),
            sidecar: ExecutionPayloadSidecar::none(),
        })
        .unwrap();

    // Verify response is INVALID
    assert!(
        outcome.outcome.is_invalid(),
        "Payload should be invalid when parent is marked invalid"
    );

    // For invalid ancestors, the latest_valid_hash behavior varies
    // We just verify it's marked as invalid
    assert!(
        outcome.outcome.latest_valid_hash.is_some() || outcome.outcome.latest_valid_hash.is_none(),
        "Latest valid hash should be set appropriately for invalid ancestor"
    );

    // Verify block 2 is now also marked as invalid
    assert!(
        test_harness.tree.state.invalid_headers.get(&hash2).is_some(),
        "Block should be added to invalid headers when parent is invalid"
    );
}

/// Test that confirms payloads received during backfill sync are buffered and reported as syncing
#[test]
fn test_on_new_payload_backfill_buffering() {
    reth_tracing::init_test_tracing();

    // Use a test data file similar to test_holesky_payload
    let s = include_str!("../../test-data/holesky/1.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 block = sealed.clone().into_block();
    let payload = ExecutionPayloadV1::from_block_unchecked(hash, &block);

    // Initialize test harness with backfill sync active
    let mut test_harness =
        TestHarness::new(HOLESKY.clone()).with_backfill_state(BackfillSyncState::Active);

    // Submit payload during backfill
    let outcome = test_harness
        .tree
        .on_new_payload(ExecutionData {
            payload: payload.into(),
            sidecar: ExecutionPayloadSidecar::none(),
        })
        .unwrap();

    // Verify response is SYNCING
    assert!(outcome.outcome.is_syncing(), "Payload should be syncing during backfill");

    // Verify the block is present in the buffer
    let hash = sealed.hash();
    let buffered_block = test_harness
        .tree
        .state
        .buffer
        .block(&hash)
        .expect("Block should be buffered during backfill sync");

    // Verify the buffered block matches what we submitted
    assert_eq!(*buffered_block, sealed, "Buffered block should match submitted payload");
}

/// Test that captures the Engine-API rule where malformed payloads report latestValidHash = None
#[test]
fn test_on_new_payload_malformed_payload() {
    reth_tracing::init_test_tracing();

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

    // Use test data
    let s = include_str!("../../test-data/holesky/1.rlp");
    let data = Bytes::from_str(s).unwrap();
    let block = Block::decode(&mut data.as_ref()).unwrap();
    let sealed = block.seal_slow();

    // Create a payload with incorrect block hash to trigger malformed validation
    let mut payload = ExecutionPayloadV1::from_block_unchecked(sealed.hash(), &sealed.into_block());

    // Corrupt the block hash - this makes the computed hash not match the provided hash
    // This will cause ensure_well_formed_payload to fail
    let wrong_hash = B256::random();
    payload.block_hash = wrong_hash;

    // Submit the malformed payload
    let outcome = test_harness
        .tree
        .on_new_payload(ExecutionData {
            payload: payload.into(),
            sidecar: ExecutionPayloadSidecar::none(),
        })
        .unwrap();

    // For malformed payloads with incorrect hash, the current implementation
    // returns SYNCING since it doesn't match computed hash
    // This test captures the current behavior to prevent regression
    assert!(
        outcome.outcome.is_syncing() || outcome.outcome.is_invalid(),
        "Malformed payload should be either syncing or invalid"
    );

    // If invalid, latestValidHash should be None per Engine API spec
    if outcome.outcome.is_invalid() {
        assert_eq!(
            outcome.outcome.latest_valid_hash, None,
            "Malformed payload must have latestValidHash = None when invalid"
        );
    }
}

/// Test different state-root job paths: the sparse-trie job and the synchronous fallback.
#[test]
fn test_state_root_strategy_paths() {
    reth_tracing::init_test_tracing();

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

    // Test multiple scenarios to ensure different state-root job paths are taken:
    // 1. The default strategy spawns the sparse-trie state-root task.
    // 2. With state-root fallback enabled, the synchronous job computes the root directly.

    let s1 = include_str!("../../test-data/holesky/1.rlp");
    let data1 = Bytes::from_str(s1).unwrap();
    let block1 = Block::decode(&mut data1.as_ref()).unwrap();
    let sealed1 = block1.seal_slow();
    let hash1 = sealed1.hash();
    let block1 = sealed1.into_block();
    let payload1 = ExecutionPayloadV1::from_block_unchecked(hash1, &block1);

    // Scenario 1: Test one strategy path
    let outcome1 = test_harness
        .tree
        .on_new_payload(ExecutionData {
            payload: payload1.into(),
            sidecar: ExecutionPayloadSidecar::none(),
        })
        .unwrap();

    assert!(
        outcome1.outcome.is_valid() || outcome1.outcome.is_syncing(),
        "First strategy path should work"
    );

    let s2 = include_str!("../../test-data/holesky/2.rlp");
    let data2 = Bytes::from_str(s2).unwrap();
    let block2 = Block::decode(&mut data2.as_ref()).unwrap();
    let sealed2 = block2.seal_slow();
    let hash2 = sealed2.hash();
    let block2 = sealed2.into_block();
    let payload2 = ExecutionPayloadV1::from_block_unchecked(hash2, &block2);

    // Scenario 2: Test different strategy path (disconnected)
    let outcome2 = test_harness
        .tree
        .on_new_payload(ExecutionData {
            payload: payload2.into(),
            sidecar: ExecutionPayloadSidecar::none(),
        })
        .unwrap();

    assert!(outcome2.outcome.is_syncing(), "Second strategy path should work");

    // This test passes if multiple state-root job scenarios work correctly:
    // - the sparse-trie job
    // - the synchronous job
}

// ================================================================================================
// VALIDATE_BLOCK_WITH_STATE TEST SUITE
// ================================================================================================
//
// This test suite exercises `validate_block_with_state` across different scenarios including:
// - Basic block validation with state root computation
// - Strategy selection based on conditions (`StateRootTask`, `Synchronous`)
// - Trie update retention and discard logic
// - Error precedence handling (consensus vs execution errors)
// - Different validation scenarios (valid, invalid consensus, invalid execution blocks)

/// Test `Synchronous` strategy when persistence is active
#[test]
fn test_validate_block_synchronous_strategy_during_persistence() {
    reth_tracing::init_test_tracing();

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

    // Set up persistence action to force `Synchronous` strategy
    use crate::tree::persistence_state::CurrentPersistenceAction;
    let persistence_action = CurrentPersistenceAction::SavingBlocks {
        highest: alloy_eips::NumHash::new(1, B256::random()),
    };
    test_harness.start_persistence_operation(persistence_action);

    // Verify persistence is active
    assert!(test_harness.is_persistence_in_progress());

    // Create valid block
    let mut block_factory = TestBlockFactory::new(MAINNET.as_ref().clone());
    let genesis_hash = MAINNET.genesis_hash();
    let valid_block = block_factory.create_valid_block(genesis_hash);

    // Test that Synchronous strategy executes during active persistence without panicking
    let _result = test_harness.validate_block_direct(valid_block);
}

/// Test multiple validation scenarios including valid, consensus-invalid, and execution-invalid
/// blocks with proper result validation
#[test]
fn test_validate_block_multiple_scenarios() {
    reth_tracing::init_test_tracing();

    // Test multiple scenarios to ensure comprehensive coverage
    let mut test_harness = ValidatorTestHarness::new(MAINNET.clone());
    let mut block_factory = TestBlockFactory::new(MAINNET.as_ref().clone());
    let genesis_hash = MAINNET.genesis_hash();

    // Scenario 1: Valid block validation (test execution, not result)
    let valid_block = block_factory.create_valid_block(genesis_hash);
    let _result1 = test_harness.validate_block_direct(valid_block);

    // Scenario 2: Block with consensus issues should be rejected
    let consensus_invalid = block_factory.create_invalid_consensus_block(genesis_hash);
    let result2 = test_harness.validate_block_direct(consensus_invalid);
    assert!(result2.is_err(), "Consensus-invalid block (invalid state root) should be rejected");

    // Scenario 3: Block with execution issues should be rejected
    let execution_invalid = block_factory.create_invalid_execution_block(genesis_hash);
    let result3 = test_harness.validate_block_direct(execution_invalid);
    assert!(result3.is_err(), "Execution-invalid block (gas limit exceeded) should be rejected");

    // Verify all validation scenarios executed without panics
    let total_calls = test_harness.validation_call_count();
    assert!(
        total_calls >= 2,
        "At least invalid block validations should have executed (got {})",
        total_calls
    );
}

/// Test suite for the `check_invalid_ancestors` method
#[cfg(test)]
mod check_invalid_ancestors_tests {
    use super::*;

    /// Test that `find_invalid_ancestor` returns None when no invalid ancestors exist
    #[test]
    fn test_find_invalid_ancestor_no_invalid() {
        reth_tracing::init_test_tracing();

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

        // Create a valid block payload
        let s = include_str!("../../test-data/holesky/1.rlp");
        let data = Bytes::from_str(s).unwrap();
        let block = Block::decode(&mut data.as_ref()).unwrap();
        let sealed = block.seal_slow();
        let payload = ExecutionData {
            payload: ExecutionPayloadV1::from_block_unchecked(sealed.hash(), &sealed.into_block())
                .into(),
            sidecar: ExecutionPayloadSidecar::none(),
        };

        // Check for invalid ancestors - should return None since none are marked invalid
        let result = test_harness.tree.find_invalid_ancestor(&payload);
        assert!(result.is_none(), "Should return None when no invalid ancestors exist");
    }

    /// Test that `find_invalid_ancestor` detects an invalid parent
    #[test]
    fn test_find_invalid_ancestor_with_invalid_parent() {
        reth_tracing::init_test_tracing();

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

        // Read block 1
        let s1 = include_str!("../../test-data/holesky/1.rlp");
        let data1 = Bytes::from_str(s1).unwrap();
        let block1 = Block::decode(&mut data1.as_ref()).unwrap();
        let sealed1 = block1.seal_slow();
        let parent1 = sealed1.parent_hash();

        // Mark block 1 as invalid
        test_harness
            .tree
            .state
            .invalid_headers
            .insert(BlockWithParent { block: sealed1.num_hash(), parent: parent1 });

        // Read block 2 which has block 1 as parent
        let s2 = include_str!("../../test-data/holesky/2.rlp");
        let data2 = Bytes::from_str(s2).unwrap();
        let block2 = Block::decode(&mut data2.as_ref()).unwrap();
        let sealed2 = block2.seal_slow();

        // Create payload for block 2
        let payload2 = ExecutionData {
            payload: ExecutionPayloadV1::from_block_unchecked(
                sealed2.hash(),
                &sealed2.into_block(),
            )
            .into(),
            sidecar: ExecutionPayloadSidecar::none(),
        };

        // Check for invalid ancestors - should detect invalid parent
        let invalid_ancestor = test_harness.tree.find_invalid_ancestor(&payload2);
        assert!(
            invalid_ancestor.is_some(),
            "Should find invalid ancestor when parent is marked as invalid"
        );

        // Now test that handling the payload with invalid ancestor returns invalid status
        let invalid = invalid_ancestor.unwrap();
        let status = test_harness.tree.handle_invalid_ancestor_payload(payload2, invalid).unwrap();
        assert!(status.is_invalid(), "Status should be invalid when parent is invalid");
    }

    /// Test genesis block handling (`parent_hash` = `B256::ZERO`)
    #[test]
    fn test_genesis_block_handling() {
        reth_tracing::init_test_tracing();

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

        // Create a genesis-like payload with parent_hash = B256::ZERO
        let mut test_block_builder = TestBlockBuilder::eth();
        let genesis_block = test_block_builder.generate_random_block(0, B256::ZERO);
        let genesis_payload = ExecutionData {
            payload: ExecutionPayloadV1::from_block_unchecked(
                genesis_block.hash(),
                &genesis_block.into_block(),
            )
            .into(),
            sidecar: ExecutionPayloadSidecar::none(),
        };

        // Check for invalid ancestors - should return None for genesis block
        let result = test_harness.tree.find_invalid_ancestor(&genesis_payload);
        assert!(result.is_none(), "Genesis block should have no invalid ancestors");
    }

    /// Test malformed payload with invalid ancestor scenario
    #[test]
    fn test_malformed_payload_with_invalid_ancestor() {
        reth_tracing::init_test_tracing();

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

        // Mark an ancestor as invalid
        let invalid_block = Block::default().seal_slow();
        test_harness.tree.state.invalid_headers.insert(BlockWithParent {
            block: invalid_block.num_hash(),
            parent: invalid_block.parent_hash(),
        });

        // Create a payload that descends from the invalid ancestor but is malformed
        let malformed_payload = create_malformed_payload_descending_from(invalid_block.hash());

        // The function should handle the malformed payload gracefully
        let invalid_ancestor = test_harness.tree.find_invalid_ancestor(&malformed_payload);
        if let Some(invalid) = invalid_ancestor {
            let status = test_harness
                .tree
                .handle_invalid_ancestor_payload(malformed_payload, invalid)
                .unwrap();
            assert!(
                status.is_invalid(),
                "Should return invalid status for malformed payload with invalid ancestor"
            );
        }
    }

    /// Test that `find_invalid_ancestor` detects the block itself in the invalid cache
    #[test]
    fn test_find_invalid_ancestor_detects_block_itself() {
        reth_tracing::init_test_tracing();

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

        // Read block 1
        let s1 = include_str!("../../test-data/holesky/1.rlp");
        let data1 = Bytes::from_str(s1).unwrap();
        let block1 = Block::decode(&mut data1.as_ref()).unwrap();
        let sealed1 = block1.seal_slow();
        let hash1 = sealed1.hash();
        let parent1 = sealed1.parent_hash();

        // Mark block 1 itself as invalid (simulates a block that failed execution)
        test_harness
            .tree
            .state
            .invalid_headers
            .insert(BlockWithParent { block: sealed1.num_hash(), parent: parent1 });

        // Create payload for block 1 (same block, sent again by CL)
        let payload1 = ExecutionData {
            payload: ExecutionPayloadV1::from_block_unchecked(hash1, &sealed1.into_block()).into(),
            sidecar: ExecutionPayloadSidecar::none(),
        };

        // find_invalid_ancestor should detect the block itself without re-execution
        let result = test_harness.tree.find_invalid_ancestor(&payload1);
        assert!(result.is_some(), "Should detect block itself in invalid headers cache");
    }

    /// Helper function to create a malformed payload that descends from a given parent
    fn create_malformed_payload_descending_from(parent_hash: B256) -> ExecutionData {
        // Create a block with invalid hash (mismatch between computed and provided hash)
        let mut test_block_builder = TestBlockBuilder::eth();
        let block = test_block_builder.generate_random_block(1, parent_hash);

        // Intentionally corrupt the block to make it malformed
        // Modify the block after creation to make validation fail
        let unsealed_block = block.unseal();

        // Create payload with wrong hash (this makes it malformed)
        let wrong_hash = B256::from([0xff; 32]);

        ExecutionData {
            payload: ExecutionPayloadV1::from_block_unchecked(wrong_hash, &unsealed_block).into(),
            sidecar: ExecutionPayloadSidecar::none(),
        }
    }
}

/// Test suite for `try_insert_payload` and `try_buffer_payload`
/// methods
#[cfg(test)]
mod payload_execution_tests {
    use super::*;

    /// Test `try_insert_payload` with different `InsertPayloadOk` variants
    #[test]
    fn test_try_insert_payload_variants() {
        reth_tracing::init_test_tracing();

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

        // Create a valid payload
        let mut test_block_builder = TestBlockBuilder::eth();
        let block = test_block_builder.generate_random_block(1, B256::ZERO);
        let payload = ExecutionData {
            payload: ExecutionPayloadV1::from_block_unchecked(block.hash(), &block.into_block())
                .into(),
            sidecar: ExecutionPayloadSidecar::none(),
        };

        // Test the function directly
        let result = test_harness.tree.try_insert_payload(payload);
        // Should handle the payload gracefully
        assert!(result.is_ok(), "Should handle valid payload without error");
    }

    /// Test `try_buffer_payload` with validation errors
    #[test]
    fn test_buffer_payload_validation_errors() {
        reth_tracing::init_test_tracing();

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

        // Create a malformed payload that will fail validation
        let malformed_payload = create_malformed_payload();

        // Test buffering during backfill sync
        let result = test_harness.tree.try_buffer_payload(malformed_payload);
        assert!(result.is_ok(), "Should handle malformed payload gracefully");
        let status = result.unwrap();
        assert!(
            status.is_invalid() || status.is_syncing(),
            "Should return invalid or syncing status for malformed payload"
        );
    }

    /// Test `try_buffer_payload` with valid payload
    #[test]
    fn test_buffer_payload_valid_payload() {
        reth_tracing::init_test_tracing();

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

        // Create a valid payload
        let mut test_block_builder = TestBlockBuilder::eth();
        let block = test_block_builder.generate_random_block(1, B256::ZERO);
        let payload = ExecutionData {
            payload: ExecutionPayloadV1::from_block_unchecked(block.hash(), &block.into_block())
                .into(),
            sidecar: ExecutionPayloadSidecar::none(),
        };

        // Test buffering during backfill sync
        let result = test_harness.tree.try_buffer_payload(payload);
        assert!(result.is_ok(), "Should handle valid payload gracefully");
        let status = result.unwrap();
        // The payload may be invalid due to missing withdrawals root, so accept either status
        assert!(
            status.is_syncing() || status.is_invalid(),
            "Should return syncing or invalid status for payload"
        );
    }

    /// Helper function to create a malformed payload
    fn create_malformed_payload() -> ExecutionData {
        // Create a payload with invalid structure that will fail validation
        let mut test_block_builder = TestBlockBuilder::eth();
        let block = test_block_builder.generate_random_block(1, B256::ZERO);

        // Modify the block to make it malformed
        let mut unsealed_block = block.unseal();

        // Corrupt the block by setting an invalid gas limit
        unsealed_block.header.gas_limit = 0;

        ExecutionData {
            payload: ExecutionPayloadV1::from_block_unchecked(
                unsealed_block.hash_slow(),
                &unsealed_block,
            )
            .into(),
            sidecar: ExecutionPayloadSidecar::none(),
        }
    }
}

/// Test suite for the refactored `on_forkchoice_updated` helper methods
#[cfg(test)]
mod forkchoice_updated_tests {
    use super::*;

    /// Test that validates the forkchoice state pre-validation logic
    #[tokio::test]
    async fn test_validate_forkchoice_state() {
        let chain_spec = MAINNET.clone();
        let mut test_harness = TestHarness::new(chain_spec);

        // Test 1: Zero head block hash should return early with invalid state
        let zero_state = ForkchoiceState {
            head_block_hash: B256::ZERO,
            safe_block_hash: B256::ZERO,
            finalized_block_hash: B256::ZERO,
        };

        let result = test_harness.tree.validate_forkchoice_state(zero_state).unwrap();
        assert!(result.is_some(), "Zero head block hash should return early");
        let outcome = result.unwrap();
        // For invalid state, we expect an error response
        assert!(matches!(outcome, OnForkChoiceUpdated { .. }));

        // Test 2: Valid state with backfill active should return syncing
        test_harness.tree.backfill_sync_state = BackfillSyncState::Active;
        let valid_state = ForkchoiceState {
            head_block_hash: B256::random(),
            safe_block_hash: B256::ZERO,
            finalized_block_hash: B256::ZERO,
        };

        let result = test_harness.tree.validate_forkchoice_state(valid_state).unwrap();
        assert!(result.is_some(), "Backfill active should return early");
        let outcome = result.unwrap();
        // We need to await the outcome to check the payload status
        let fcu_result = outcome.await.unwrap();
        assert!(fcu_result.payload_status.is_syncing());

        // Test 3: Valid state with idle backfill should continue processing
        test_harness.tree.backfill_sync_state = BackfillSyncState::Idle;
        let valid_state = ForkchoiceState {
            head_block_hash: B256::random(),
            safe_block_hash: B256::ZERO,
            finalized_block_hash: B256::ZERO,
        };

        let result = test_harness.tree.validate_forkchoice_state(valid_state).unwrap();
        assert!(result.is_none(), "Valid state should continue processing");
    }

    /// Test that verifies canonical head handling
    #[tokio::test]
    async fn test_handle_canonical_head() {
        let chain_spec = MAINNET.clone();
        let mut test_harness = TestHarness::new(chain_spec);

        // Create test blocks
        let blocks: Vec<_> = test_harness.block_builder.get_executed_blocks(0..3).collect();
        test_harness = test_harness.with_blocks(blocks);

        let canonical_head = test_harness.tree.state.tree_state.canonical_block_hash();

        // Test 1: Head is already canonical, no payload attributes
        let state = ForkchoiceState {
            head_block_hash: canonical_head,
            safe_block_hash: B256::ZERO,
            finalized_block_hash: B256::ZERO,
        };

        let result = test_harness.tree.handle_canonical_head(state, &None).unwrap();
        assert!(result.is_some(), "Should return outcome for canonical head");
        let outcome = result.unwrap();
        let fcu_result = outcome.outcome.await.unwrap();
        assert!(fcu_result.payload_status.is_valid());

        // Test 2: Head is not canonical - should return None to continue processing
        let non_canonical_state = ForkchoiceState {
            head_block_hash: B256::random(),
            safe_block_hash: B256::ZERO,
            finalized_block_hash: B256::ZERO,
        };

        let result = test_harness.tree.handle_canonical_head(non_canonical_state, &None).unwrap();
        assert!(result.is_none(), "Non-canonical head should return None");
    }

    /// Test that verifies chain update application
    #[tokio::test]
    async fn test_apply_chain_update() {
        let chain_spec = MAINNET.clone();
        let mut test_harness = TestHarness::new(chain_spec);

        // Create a chain of blocks
        let blocks: Vec<_> = test_harness.block_builder.get_executed_blocks(0..5).collect();
        test_harness = test_harness.with_blocks(blocks.clone());

        let new_head = blocks[2].recovered_block().hash();

        // Test 1: Apply chain update to a new head
        let state = ForkchoiceState {
            head_block_hash: new_head,
            safe_block_hash: B256::ZERO,
            finalized_block_hash: B256::ZERO,
        };

        let result = test_harness.tree.apply_chain_update(state, &None).unwrap();
        assert!(result.is_some(), "Should apply chain update for new head");
        let outcome = result.unwrap();
        let fcu_result = outcome.outcome.await.unwrap();
        assert!(fcu_result.payload_status.is_valid());

        // Test 2: Try to apply chain update to missing block
        let missing_state = ForkchoiceState {
            head_block_hash: B256::random(),
            safe_block_hash: B256::ZERO,
            finalized_block_hash: B256::ZERO,
        };

        let result = test_harness.tree.apply_chain_update(missing_state, &None).unwrap();
        assert!(result.is_none(), "Missing block should return None");
    }

    /// Test that verifies missing block handling
    #[tokio::test]
    async fn test_handle_missing_block() {
        let chain_spec = MAINNET.clone();
        let test_harness = TestHarness::new(chain_spec);

        let state = ForkchoiceState {
            head_block_hash: B256::random(),
            safe_block_hash: B256::ZERO,
            finalized_block_hash: B256::ZERO,
        };

        let result = test_harness.tree.handle_missing_block(state).unwrap();

        // Should return syncing status with download event
        let fcu_result = result.outcome.await.unwrap();
        assert!(fcu_result.payload_status.is_syncing());
        assert!(result.event.is_some());

        if let Some(TreeEvent::Download(download_request)) = result.event {
            match download_request {
                DownloadRequest::BlockSet(block_set) => {
                    assert_eq!(block_set.len(), 1);
                }
                _ => panic!("Expected single block download request"),
            }
        }
    }

    /// Test the complete `on_forkchoice_updated` flow with all helper methods
    #[tokio::test]
    async fn test_on_forkchoice_updated_integration() {
        reth_tracing::init_test_tracing();

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

        // Create test blocks
        let blocks: Vec<_> = test_harness.block_builder.get_executed_blocks(0..3).collect();
        test_harness = test_harness.with_blocks(blocks.clone());

        let canonical_head = test_harness.tree.state.tree_state.canonical_block_hash();

        // Test Case 1: FCU to existing canonical head
        let state = ForkchoiceState {
            head_block_hash: canonical_head,
            safe_block_hash: canonical_head,
            finalized_block_hash: canonical_head,
        };

        let result = test_harness.tree.on_forkchoice_updated(state, None).unwrap();
        let fcu_result = result.outcome.await.unwrap();
        assert!(fcu_result.payload_status.is_valid());

        // Test Case 2: FCU to missing block
        let missing_state = ForkchoiceState {
            head_block_hash: B256::random(),
            safe_block_hash: B256::ZERO,
            finalized_block_hash: B256::ZERO,
        };

        let result = test_harness.tree.on_forkchoice_updated(missing_state, None).unwrap();
        let fcu_result = result.outcome.await.unwrap();
        assert!(fcu_result.payload_status.is_syncing());
        assert!(result.event.is_some(), "Should trigger download event for missing block");

        // Test Case 3: FCU during backfill sync
        test_harness.tree.backfill_sync_state = BackfillSyncState::Active;
        let state = ForkchoiceState {
            head_block_hash: canonical_head,
            safe_block_hash: B256::ZERO,
            finalized_block_hash: B256::ZERO,
        };

        let result = test_harness.tree.on_forkchoice_updated(state, None).unwrap();
        let fcu_result = result.outcome.await.unwrap();
        assert!(fcu_result.payload_status.is_syncing(), "Should return syncing during backfill");
    }

    /// Test edge case: FCU with invalid ancestor
    #[tokio::test]
    async fn test_fcu_with_invalid_ancestor() {
        let chain_spec = MAINNET.clone();
        let mut test_harness = TestHarness::new(chain_spec);

        // Mark a block as invalid
        let invalid_block_hash = B256::random();
        test_harness.tree.state.invalid_headers.insert(BlockWithParent {
            block: NumHash::new(1, invalid_block_hash),
            parent: B256::ZERO,
        });

        // Test FCU that points to a descendant of the invalid block
        // This is a bit tricky to test directly, but we can verify the check_invalid_ancestor
        // method
        let result = test_harness.tree.check_invalid_ancestor(invalid_block_hash).unwrap();
        assert!(result.is_some(), "Should detect invalid ancestor");
    }

    /// Test `OpStack` specific behavior with canonical head
    #[tokio::test]
    async fn test_opstack_canonical_head_behavior() {
        let chain_spec = MAINNET.clone();
        let mut test_harness = TestHarness::new(chain_spec);

        // Set engine kind to OpStack
        test_harness.tree.engine_kind = EngineApiKind::OpStack;

        // Create test blocks
        let blocks: Vec<_> = test_harness.block_builder.get_executed_blocks(0..3).collect();
        test_harness = test_harness.with_blocks(blocks);

        let canonical_head = test_harness.tree.state.tree_state.canonical_block_hash();

        // For OpStack, even if head is already canonical, we should still process payload
        // attributes
        let state = ForkchoiceState {
            head_block_hash: canonical_head,
            safe_block_hash: B256::ZERO,
            finalized_block_hash: B256::ZERO,
        };

        let result = test_harness.tree.handle_canonical_head(state, &None).unwrap();
        assert!(result.is_some(), "OpStack should handle canonical head");
    }

    #[test]
    fn test_update_reorg_metrics() {
        let chain_spec = MAINNET.clone();
        let test_harness = TestHarness::new(chain_spec);

        let seal_header = |number: u64| {
            SealedHeader::seal_slow(alloy_consensus::Header { number, ..Default::default() })
        };

        // Set finalized=30, safe=50 to test all three commitment levels
        test_harness.tree.canonical_in_memory_state.set_finalized(seal_header(30));
        test_harness.tree.canonical_in_memory_state.set_safe(seal_header(50));

        // Reorg at block 20 (below finalized=30) -> finalized reorg
        test_harness.tree.update_reorg_metrics(5, Some(NumHash::new(20, B256::random())));

        // Reorg at block 40 (below safe=50, above finalized=30) -> safe reorg
        test_harness.tree.update_reorg_metrics(3, Some(NumHash::new(40, B256::random())));

        // Reorg at block 60 (above safe=50) -> head reorg
        test_harness.tree.update_reorg_metrics(2, Some(NumHash::new(60, B256::random())));
    }

    /// Test that engine termination persists all blocks and signals completion.
    #[test]
    fn test_engine_termination_with_everything_persisted() {
        let chain_spec = MAINNET.clone();
        let mut test_block_builder = TestBlockBuilder::eth().with_chain_spec((*chain_spec).clone());

        // Create 10 blocks to persist
        let blocks: Vec<_> = test_block_builder.get_executed_blocks(1..11).collect();
        let canonical_tip = blocks.last().unwrap().recovered_block().number;
        let test_harness = TestHarness::new(chain_spec).with_blocks(blocks);

        // Create termination channel
        let (terminate_tx, mut terminate_rx) = oneshot::channel();

        let to_tree_tx = test_harness.to_tree_tx.clone();
        let action_rx = test_harness.action_rx;

        // Spawn tree in background thread
        spawn_os_thread("engine", || test_harness.tree.run());

        // Send terminate request
        to_tree_tx
            .send(FromEngine::Event(FromOrchestrator::Terminate { tx: terminate_tx }))
            .unwrap();

        // Handle persistence actions until termination completes
        let mut last_persisted_number = 0;
        loop {
            if terminate_rx.try_recv().is_ok() {
                break;
            }

            if let Ok(PersistenceAction::SaveBlocks(input, sender)) =
                action_rx.recv_timeout(std::time::Duration::from_millis(100))
            {
                let last = input.last_block();
                last_persisted_number = last.number;
                sender
                    .send(PersistenceResult {
                        last_block: last,
                        last_state_trie_block: last,
                        commit_duration: Some(Duration::ZERO),
                    })
                    .unwrap();
            }
        }

        // Ensure we persisted right to the tip
        assert_eq!(last_persisted_number, canonical_tip);
    }

    #[test]
    fn test_engine_termination_catches_up_state_trie_at_database_tip() {
        let chain_spec = MAINNET.clone();
        let mut test_block_builder = TestBlockBuilder::eth().with_chain_spec((*chain_spec).clone());
        let blocks: Vec<_> = test_block_builder.get_executed_blocks(1..11).collect();
        let database_tip = blocks.last().unwrap().recovered_block().num_hash();
        let state_trie_tip = blocks[4].recovered_block().num_hash();
        let mut test_harness = TestHarness::new(chain_spec).with_blocks(blocks);
        test_harness.tree.persistence_state.last_persisted_block = database_tip;
        test_harness.tree.persistence_state.last_state_trie_persisted_block = state_trie_tip;

        let (terminate_tx, terminate_rx) = oneshot::channel();
        let to_tree_tx = test_harness.to_tree_tx.clone();
        let action_rx = test_harness.action_rx;
        spawn_os_thread("engine", || test_harness.tree.run());

        to_tree_tx
            .send(FromEngine::Event(FromOrchestrator::Terminate { tx: terminate_tx }))
            .unwrap();

        let action = action_rx.recv_timeout(Duration::from_secs(10)).unwrap();
        let PersistenceAction::SaveBlocks(input, sender) = action else {
            panic!("expected state/trie catch-up save, got {action:?}")
        };
        assert_eq!(input.new_db_tip(), database_tip.number);
        assert_eq!(input.new_partial_state_trie(), database_tip.number);
        assert!(input.persist_rest_blocks().is_empty());

        sender
            .send(PersistenceResult {
                last_block: database_tip,
                last_state_trie_block: database_tip,
                commit_duration: Some(Duration::ZERO),
            })
            .unwrap();

        terminate_rx.blocking_recv().unwrap();
    }
}

/// Tests that `on_valid_downloaded_block` triggers a download for the actual head block when
/// the block matches a non-head sync target (safe or finalized).
///
/// This exercises the exact code path fixed in `on_downloaded_block`: after `insert_block`
/// returns `Inserted(Valid)`, `on_valid_downloaded_block` checks `sync_target.contains()`.
/// If the block is NOT the head, it should make canonical inline and emit a `Download`
/// event for the head — rather than returning `MakeCanonical` which would stop the download
/// pipeline.
///
/// Reproduces the hive test failure:
///   "Sync after 2 blocks - Withdrawals on Block 2 - Multiple Withdrawal Accounts -
///    No Transactions: Timeout while waiting for secondary client to sync"
#[test]
fn test_on_valid_downloaded_non_head_sync_target_continues_to_head() {
    reth_tracing::init_test_tracing();

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

    // Build blocks: genesis (0) and safe block (1).
    let blocks: Vec<_> = test_harness.block_builder.get_executed_blocks(0..2).collect();
    let genesis = &blocks[0];
    let safe_block = &blocks[1];

    // Insert genesis and safe block into the tree. The safe block must be in the tree
    // for `make_canonical` to succeed inside `on_valid_downloaded_block`.
    test_harness = test_harness.with_blocks(vec![genesis.clone(), safe_block.clone()]);

    let genesis_hash = genesis.recovered_block().hash();
    let safe_hash = safe_block.recovered_block().hash();
    let head_hash = B256::random(); // head block is unknown — hasn't been downloaded yet

    // Reset canonical head to genesis so the safe block is in tree but not yet canonical.
    test_harness.tree.state.tree_state.set_canonical_head(genesis.recovered_block().num_hash());

    // Set the forkchoice tracker to SYNCING with head != safe.
    let fcu_state = ForkchoiceState {
        head_block_hash: head_hash,
        safe_block_hash: safe_hash,
        finalized_block_hash: genesis_hash,
    };
    test_harness
        .tree
        .state
        .forkchoice_state_tracker
        .set_latest(fcu_state, ForkchoiceStatus::Syncing);

    // Call on_valid_downloaded_block — this is called by on_downloaded_block after
    // insert_block returns Inserted(Valid).
    let safe_num_hash = safe_block.recovered_block().num_hash();
    let result = test_harness.tree.on_valid_downloaded_block(safe_num_hash).unwrap();

    // With the fix: the engine makes safe canonical inline, then emits Download for head.
    // Without the fix: it would return MakeCanonical{safe_hash} and never download head.
    match result {
        Some(TreeEvent::Download(DownloadRequest::BlockSet(hashes))) => {
            assert!(
                hashes.contains(&head_hash),
                "Expected download for head block {head_hash}, got {hashes:?}"
            );
        }
        Some(TreeEvent::TreeAction(TreeAction::MakeCanonical { sync_target_head })) => {
            panic!(
                "BUG: returned MakeCanonical for non-head block {sync_target_head} \
                 instead of downloading the actual head {head_hash}"
            );
        }
        other => panic!("Expected Download event for head block, got: {other:?}"),
    }

    // Verify the safe block was made canonical.
    assert_eq!(
        test_harness.tree.state.tree_state.canonical_block_hash(),
        safe_hash,
        "Safe block should be canonical after on_valid_downloaded_block"
    );
}

/// Tests that `on_valid_downloaded_block` returns `MakeCanonical` when the downloaded block
/// IS the sync target head (the normal non-buggy path).
#[test]
fn test_on_valid_downloaded_head_sync_target_returns_make_canonical() {
    reth_tracing::init_test_tracing();

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

    let blocks: Vec<_> = test_harness.block_builder.get_executed_blocks(0..2).collect();
    let genesis = &blocks[0];
    let head_block = &blocks[1];

    test_harness = test_harness.with_blocks(vec![genesis.clone(), head_block.clone()]);

    let genesis_hash = genesis.recovered_block().hash();
    let head_hash = head_block.recovered_block().hash();

    // Reset canonical head to genesis.
    test_harness.tree.state.tree_state.set_canonical_head(genesis.recovered_block().num_hash());

    // Set the forkchoice tracker: head == the downloaded block.
    let fcu_state = ForkchoiceState {
        head_block_hash: head_hash,
        safe_block_hash: head_hash,
        finalized_block_hash: genesis_hash,
    };
    test_harness
        .tree
        .state
        .forkchoice_state_tracker
        .set_latest(fcu_state, ForkchoiceStatus::Syncing);

    let head_num_hash = head_block.recovered_block().num_hash();
    let result = test_harness.tree.on_valid_downloaded_block(head_num_hash).unwrap();

    // When the downloaded block IS the head, should return MakeCanonical.
    match result {
        Some(TreeEvent::TreeAction(TreeAction::MakeCanonical { sync_target_head })) => {
            assert_eq!(sync_target_head, head_hash);
        }
        other => panic!("Expected MakeCanonical for head block, got: {other:?}"),
    }
}

/// Tests that canonicalizing a downloaded sync target head also applies the tracked finalized
/// block from the original `SYNCING` forkchoice state.
#[test]
fn test_canonicalizing_downloaded_sync_target_head_updates_finalized() {
    reth_tracing::init_test_tracing();

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

    let blocks: Vec<_> = test_harness.block_builder.get_executed_blocks(0..3).collect();
    let genesis = &blocks[0];
    let finalized_block = &blocks[1];
    let head_block = &blocks[2];

    test_harness = test_harness.with_blocks(vec![
        genesis.clone(),
        finalized_block.clone(),
        head_block.clone(),
    ]);

    let finalized_num_hash = finalized_block.recovered_block().num_hash();
    let head_num_hash = head_block.recovered_block().num_hash();

    test_harness.tree.state.tree_state.set_canonical_head(genesis.recovered_block().num_hash());

    let fcu_state = ForkchoiceState {
        head_block_hash: head_num_hash.hash,
        safe_block_hash: head_num_hash.hash,
        finalized_block_hash: finalized_num_hash.hash,
    };
    test_harness
        .tree
        .state
        .forkchoice_state_tracker
        .set_latest(fcu_state, ForkchoiceStatus::Syncing);

    let event = test_harness
        .tree
        .on_valid_downloaded_block(head_num_hash)
        .unwrap()
        .expect("expected canonicalization event for sync target head");

    test_harness.tree.on_tree_event(event).unwrap();

    assert_eq!(test_harness.tree.state.tree_state.canonical_block_hash(), head_num_hash.hash);
    assert_eq!(
        test_harness.tree.canonical_in_memory_state.get_finalized_num_hash(),
        Some(finalized_num_hash),
        "Finalized block from the syncing FCU should be applied once the head becomes canonical"
    );
    assert_eq!(
        test_harness.tree.canonical_in_memory_state.get_safe_num_hash(),
        Some(head_num_hash),
        "Safe block from the syncing FCU should be applied once the head becomes canonical"
    );
    assert_eq!(
        test_harness.tree.state.forkchoice_state_tracker.last_valid_state(),
        Some(fcu_state)
    );
    assert!(test_harness.tree.state.forkchoice_state_tracker.sync_target_state().is_none());
}

// --- Backfill target selection tests ---
//
// Cover `backfill_target_hash` and its consumer `backfill_sync_target`, exercised end-to-end
// via `on_disconnected_downloaded_block`. The OP Stack branch targets head; the Ethereum
// branch targets finalized. When the CL has no finalized hash yet, the optimistic fallback to
// head is handled by `backfill_sync_target`.

#[test]
fn test_backfill_target_hash_eth_returns_finalized() {
    let test_harness = TestHarness::new(MAINNET.clone());
    let head = B256::from([0xAA; 32]);
    let finalized = B256::from([0xBB; 32]);
    let state = ForkchoiceState {
        head_block_hash: head,
        safe_block_hash: B256::ZERO,
        finalized_block_hash: finalized,
    };

    assert_eq!(test_harness.tree.backfill_target_hash(state), finalized);
}

#[test]
fn test_backfill_target_hash_eth_returns_zero_finalized() {
    let test_harness = TestHarness::new(MAINNET.clone());
    let head = B256::from([0xAA; 32]);
    let state = ForkchoiceState {
        head_block_hash: head,
        safe_block_hash: B256::ZERO,
        finalized_block_hash: B256::ZERO,
    };

    assert_eq!(test_harness.tree.backfill_target_hash(state), B256::ZERO);
}

#[test]
fn test_backfill_target_hash_opstack_returns_head() {
    let mut test_harness = TestHarness::new(MAINNET.clone());
    test_harness.tree.engine_kind = EngineApiKind::OpStack;
    let head = B256::from([0xAA; 32]);
    let finalized = B256::from([0xBB; 32]);
    let state = ForkchoiceState {
        head_block_hash: head,
        safe_block_hash: B256::ZERO,
        finalized_block_hash: finalized,
    };

    // OP Stack: finalized can lag far behind the canonical tip; target head regardless.
    assert_eq!(test_harness.tree.backfill_target_hash(state), head);
}

#[test]
fn test_backfill_sync_target_without_sync_state_returns_none() {
    let test_harness = TestHarness::new(MAINNET.clone());
    assert_eq!(
        test_harness.tree.backfill_sync_target(0, MIN_BLOCKS_FOR_PIPELINE_RUN + 100, None),
        None
    );
}

/// On OP Stack, a disconnected downloaded block whose missing parent is far ahead of the
/// canonical tip should trigger a backfill targeting `head_block_hash`, not finalized.
#[test]
fn test_on_disconnected_downloaded_block_opstack_targets_head() {
    let mut test_harness = TestHarness::new(MAINNET.clone());
    test_harness.tree.engine_kind = EngineApiKind::OpStack;

    let head_hash = B256::from([0xAA; 32]);
    let finalized_hash = B256::from([0xBB; 32]);
    test_harness.tree.state.forkchoice_state_tracker.set_latest(
        ForkchoiceState {
            head_block_hash: head_hash,
            safe_block_hash: B256::ZERO,
            finalized_block_hash: finalized_hash,
        },
        ForkchoiceStatus::Syncing,
    );

    let canonical_head = BlockNumHash::new(0, B256::ZERO);
    let downloaded_block =
        BlockNumHash::new(MIN_BLOCKS_FOR_PIPELINE_RUN + 100, B256::from([0xCC; 32]));
    let missing_parent =
        BlockNumHash::new(MIN_BLOCKS_FOR_PIPELINE_RUN + 99, B256::from([0xDD; 32]));

    let event = test_harness.tree.on_disconnected_downloaded_block(
        downloaded_block,
        missing_parent,
        canonical_head,
    );

    match event {
        Some(TreeEvent::BackfillAction(BackfillAction::Start(target))) => {
            assert_eq!(
                target.sync_target(),
                Some(head_hash),
                "OP Stack backfill should target head, not finalized"
            );
        }
        other => panic!("Expected BackfillAction(Start), got: {other:?}"),
    }
}

/// On Ethereum, a disconnected downloaded block whose missing parent is far ahead of the
/// canonical tip should trigger a backfill targeting `finalized_block_hash`.
#[test]
fn test_on_disconnected_downloaded_block_eth_targets_finalized() {
    let mut test_harness = TestHarness::new(MAINNET.clone());

    let head_hash = B256::from([0xAA; 32]);
    let finalized_hash = B256::from([0xBB; 32]);
    test_harness.tree.state.forkchoice_state_tracker.set_latest(
        ForkchoiceState {
            head_block_hash: head_hash,
            safe_block_hash: B256::ZERO,
            finalized_block_hash: finalized_hash,
        },
        ForkchoiceStatus::Syncing,
    );

    let canonical_head = BlockNumHash::new(0, B256::ZERO);
    let downloaded_block =
        BlockNumHash::new(MIN_BLOCKS_FOR_PIPELINE_RUN + 100, B256::from([0xCC; 32]));
    let missing_parent =
        BlockNumHash::new(MIN_BLOCKS_FOR_PIPELINE_RUN + 99, B256::from([0xDD; 32]));

    let event = test_harness.tree.on_disconnected_downloaded_block(
        downloaded_block,
        missing_parent,
        canonical_head,
    );

    match event {
        Some(TreeEvent::BackfillAction(BackfillAction::Start(target))) => {
            assert_eq!(
                target.sync_target(),
                Some(finalized_hash),
                "Ethereum backfill should target finalized"
            );
        }
        other => panic!("Expected BackfillAction(Start), got: {other:?}"),
    }
}

/// On Ethereum, a zero finalized hash means optimistic sync. The helper still selects finalized,
/// but the sync-target builder falls back to `head_block_hash`.
#[test]
fn test_on_disconnected_downloaded_block_eth_zero_finalized_targets_head() {
    let mut test_harness = TestHarness::new(MAINNET.clone());

    let head_hash = B256::from([0xAA; 32]);
    test_harness.tree.state.forkchoice_state_tracker.set_latest(
        ForkchoiceState {
            head_block_hash: head_hash,
            safe_block_hash: B256::ZERO,
            finalized_block_hash: B256::ZERO,
        },
        ForkchoiceStatus::Syncing,
    );

    let canonical_head = BlockNumHash::new(0, B256::ZERO);
    let downloaded_block =
        BlockNumHash::new(MIN_BLOCKS_FOR_PIPELINE_RUN + 100, B256::from([0xCC; 32]));
    let missing_parent =
        BlockNumHash::new(MIN_BLOCKS_FOR_PIPELINE_RUN + 99, B256::from([0xDD; 32]));

    let event = test_harness.tree.on_disconnected_downloaded_block(
        downloaded_block,
        missing_parent,
        canonical_head,
    );

    match event {
        Some(TreeEvent::BackfillAction(BackfillAction::Start(target))) => {
            assert_eq!(
                target.sync_target(),
                Some(head_hash),
                "Ethereum optimistic backfill should target head when finalized is zero"
            );
        }
        other => panic!("Expected BackfillAction(Start), got: {other:?}"),
    }
}

/// Verifies that the post-backfill recheck path in `on_backfill_sync_finished` retriggers a
/// new backfill targeting whichever block `backfill_target_hash` resolves to — head on OP
/// Stack, finalized on Ethereum — when that block is buffered far ahead of where the
/// just-finished pipeline landed.
async fn assert_post_backfill_recheck_retriggers_to_buffered_target(engine_kind: EngineApiKind) {
    reth_tracing::init_test_tracing();

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

    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;

    // Long unsynced chain. The last block is what the helper should resolve to.
    let main_chain = test_harness
        .block_builder
        .create_fork(base_chain[0].recovered_block(), MIN_BLOCKS_FOR_PIPELINE_RUN + 50);
    let target_block = main_chain.last().unwrap().clone();
    let target_hash = target_block.hash();

    // Buffer the target block — the recheck looks up the helper's resolved hash in the buffer
    // to decide whether to retrigger.
    test_harness.tree.state.buffer.insert_block(target_block.clone_sealed_block());

    // Place the buffered hash in the FCU slot the helper picks for this chain type, and put
    // an unrelated hash (not in buffer) in the other slot to keep the two slots distinct.
    let other_hash = B256::from([0xFF; 32]);
    let (head_block_hash, finalized_block_hash) = if engine_kind.is_opstack() {
        (target_hash, other_hash)
    } else {
        (other_hash, target_hash)
    };
    test_harness.tree.state.forkchoice_state_tracker.set_latest(
        ForkchoiceState { head_block_hash, safe_block_hash: head_block_hash, finalized_block_hash },
        ForkchoiceStatus::Syncing,
    );

    // Simulate backfill finishing far below the buffered target (gap > threshold).
    let backfill_finished_block_number = MIN_BLOCKS_FOR_PIPELINE_RUN + 1;
    let backfill_tip_block = main_chain[(backfill_finished_block_number - 1) as usize].clone();
    test_harness.provider.add_block(backfill_tip_block.hash(), backfill_tip_block.into_block());
    let backfill_finished = FromOrchestrator::BackfillSyncFinished(ControlFlow::Continue {
        block_number: backfill_finished_block_number,
    });
    let _ = test_harness.tree.on_engine_message(FromEngine::Event(backfill_finished)).unwrap();

    let event = test_harness.from_tree_rx.recv().await.unwrap();
    match event {
        EngineApiEvent::BackfillAction(BackfillAction::Start(emitted_target)) => {
            assert_eq!(
                emitted_target.sync_target(),
                Some(target_hash),
                "post-backfill recheck should retrigger backfill to the buffered target"
            );
        }
        _ => panic!("Expected BackfillAction(Start), got: {event:#?}"),
    }
}

#[tokio::test]
async fn test_on_backfill_sync_finished_opstack_retriggers_backfill_to_buffered_head() {
    assert_post_backfill_recheck_retriggers_to_buffered_target(EngineApiKind::OpStack).await;
}

#[tokio::test]
async fn test_on_backfill_sync_finished_eth_retriggers_backfill_to_buffered_finalized() {
    assert_post_backfill_recheck_retriggers_to_buffered_target(EngineApiKind::Ethereum).await;
}
