reth_rpc_engine_api/

engine_api.rs

use crate::{
    capabilities::EngineCapabilities, metrics::EngineApiMetrics, EngineApiError, EngineApiResult,
};
use alloy_eips::{
    eip1898::BlockHashOrNumber,
    eip4844::{BlobAndProofV1, BlobAndProofV2, BlobCellsAndProofsV1},
    eip4895::Withdrawals,
    eip7685::RequestsOrHash,
};
use alloy_primitives::{BlockHash, BlockNumber, Bytes, Sealable, B128, B256, U64};
use alloy_rpc_types_engine::{
    BogotaPayloadFields, CancunPayloadFields, ClientVersionV1, ExecutionData,
    ExecutionPayloadBodiesV1, ExecutionPayloadBodiesV2, ExecutionPayloadBodyV1,
    ExecutionPayloadBodyV2, ExecutionPayloadInputV2, ExecutionPayloadSidecar, ExecutionPayloadV1,
    ExecutionPayloadV3, ExecutionPayloadV4, ForkchoiceState, ForkchoiceUpdated,
    ForkchoiceUpdatedResponseV2, PayloadId, PayloadStatus, PayloadStatusV2, PraguePayloadFields,
    MAX_BYTES_PER_INCLUSION_LIST,
};
use async_trait::async_trait;
use jsonrpsee_core::{server::RpcModule, RpcResult};
use reth_chainspec::EthereumHardforks;
use reth_engine_primitives::{ConsensusEngineHandle, EngineApiValidator, EngineTypes};
use reth_network_api::{CellCustody, NetworkInfo};
use reth_payload_builder::PayloadStore;
use reth_payload_primitives::{
    validate_payload_timestamp, EngineApiMessageVersion, MessageValidationKind,
    PayloadOrAttributes, PayloadTypes,
};
use reth_primitives_traits::{AlloyBlockHeader, Block, BlockBody};
use reth_rpc_api::{EngineApiServer, IntoEngineApiRpcModule};
use reth_storage_api::{BalProvider, BlockReader, HeaderProvider, StateProviderFactory};
use reth_tasks::Runtime;
use reth_transaction_pool::{BestTransactions, TransactionPool};
use std::{
    sync::Arc,
    time::{Instant, SystemTime},
};
use tokio::sync::oneshot;
use tracing::{debug, trace, warn};

/// The Engine API response sender.
pub type EngineApiSender<Ok> = oneshot::Sender<EngineApiResult<Ok>>;

/// The upper limit for payload bodies request.
const MAX_PAYLOAD_BODIES_LIMIT: u64 = 1024;

/// The upper limit for blobs in `engine_getBlobsVx`.
const MAX_BLOB_LIMIT: usize = 128;

/// The Engine API implementation that grants the Consensus layer access to data and
/// functions in the Execution layer that are crucial for the consensus process.
///
/// This type is generic over [`EngineTypes`] and intended to be used as the entrypoint for engine
/// API processing. It can be reused by other non L1 engine APIs that deviate from the L1 spec but
/// are still follow the engine API model.
///
/// ## Implementers
///
/// Implementing support for an engine API jsonrpsee RPC handler is done by defining the engine API
/// server trait and implementing it on a type that can either wrap this [`EngineApi`] type or
/// use a custom [`EngineTypes`] implementation if it mirrors ethereum's versioned engine API
/// endpoints (e.g. opstack).
/// See also [`EngineApiServer`] implementation for this type which is the
/// L1 implementation.
pub struct EngineApi<Provider, PayloadT: PayloadTypes, Pool, Validator, ChainSpec> {
    inner: Arc<EngineApiInner<Provider, PayloadT, Pool, Validator, ChainSpec>>,
}

impl<Provider, PayloadT: PayloadTypes, Pool, Validator, ChainSpec>
    EngineApi<Provider, PayloadT, Pool, Validator, ChainSpec>
{
    /// Returns the configured chainspec.
    pub fn chain_spec(&self) -> &Arc<ChainSpec> {
        &self.inner.chain_spec
    }

    /// Returns the configured client version.
    pub fn client_version(&self) -> &ClientVersionV1 {
        &self.inner.client
    }
}

impl<Provider, PayloadT, Pool, Validator, ChainSpec>
    EngineApi<Provider, PayloadT, Pool, Validator, ChainSpec>
where
    Provider: HeaderProvider + BlockReader + StateProviderFactory + BalProvider + 'static,
    PayloadT: PayloadTypes,
    Pool: TransactionPool + 'static,
    Validator: EngineApiValidator<PayloadT>,
    ChainSpec: EthereumHardforks + Send + Sync + 'static,
{
    /// Create new instance of [`EngineApi`].
    #[expect(clippy::too_many_arguments)]
    pub fn new(
        provider: Provider,
        chain_spec: Arc<ChainSpec>,
        beacon_consensus: ConsensusEngineHandle<PayloadT>,
        payload_store: PayloadStore<PayloadT>,
        tx_pool: Pool,
        task_spawner: Runtime,
        client: ClientVersionV1,
        capabilities: EngineCapabilities,
        validator: Validator,
        accept_execution_requests_hash: bool,
        network: impl NetworkInfo + 'static,
    ) -> Self {
        let cell_custody = network.cell_custody().clone();
        let is_syncing = Arc::new(move || network.is_syncing());
        let inner = Arc::new(EngineApiInner {
            provider,
            chain_spec,
            beacon_consensus,
            payload_store,
            task_spawner,
            metrics: EngineApiMetrics::default(),
            client,
            capabilities,
            tx_pool,
            validator,
            accept_execution_requests_hash,
            cell_custody,
            is_syncing,
        });
        Self { inner }
    }

    /// Fetches the client version.
    pub fn get_client_version_v1(
        &self,
        _client: ClientVersionV1,
    ) -> EngineApiResult<Vec<ClientVersionV1>> {
        Ok(vec![self.inner.client.clone()])
    }

    /// Fetches the timestamp of the payload with the given id.
    async fn get_payload_timestamp(&self, payload_id: PayloadId) -> EngineApiResult<u64> {
        Ok(self
            .inner
            .payload_store
            .payload_timestamp(payload_id)
            .await
            .ok_or(EngineApiError::UnknownPayload)??)
    }

    /// See also <https://github.com/ethereum/execution-apis/blob/3d627c95a4d3510a8187dd02e0250ecb4331d27e/src/engine/paris.md#engine_newpayloadv1>
    /// Caution: This should not accept the `withdrawals` field
    pub async fn new_payload_v1(
        &self,
        payload: PayloadT::ExecutionData,
    ) -> EngineApiResult<PayloadStatus> {
        let payload_or_attrs = PayloadOrAttributes::<
            '_,
            PayloadT::ExecutionData,
            PayloadT::PayloadAttributes,
        >::from_execution_payload(&payload);

        self.inner
            .validator
            .validate_version_specific_fields(EngineApiMessageVersion::V1, payload_or_attrs)?;

        Ok(self.inner.beacon_consensus.new_payload(payload).await?)
    }

    /// Metered version of `new_payload_v1`.
    pub async fn new_payload_v1_metered(
        &self,
        payload: PayloadT::ExecutionData,
    ) -> EngineApiResult<PayloadStatus> {
        let start = Instant::now();
        let res = Self::new_payload_v1(self, payload).await;
        let elapsed = start.elapsed();
        self.inner.metrics.latency.new_payload_v1.record(elapsed);
        res
    }

    /// See also <https://github.com/ethereum/execution-apis/blob/584905270d8ad665718058060267061ecfd79ca5/src/engine/shanghai.md#engine_newpayloadv2>
    pub async fn new_payload_v2(
        &self,
        payload: PayloadT::ExecutionData,
    ) -> EngineApiResult<PayloadStatus> {
        let payload_or_attrs = PayloadOrAttributes::<
            '_,
            PayloadT::ExecutionData,
            PayloadT::PayloadAttributes,
        >::from_execution_payload(&payload);
        self.inner
            .validator
            .validate_version_specific_fields(EngineApiMessageVersion::V2, payload_or_attrs)?;
        Ok(self.inner.beacon_consensus.new_payload(payload).await?)
    }

    /// Metered version of `new_payload_v2`.
    pub async fn new_payload_v2_metered(
        &self,
        payload: PayloadT::ExecutionData,
    ) -> EngineApiResult<PayloadStatus> {
        let start = Instant::now();
        let res = Self::new_payload_v2(self, payload).await;
        let elapsed = start.elapsed();
        self.inner.metrics.latency.new_payload_v2.record(elapsed);
        res
    }

    /// See also <https://github.com/ethereum/execution-apis/blob/fe8e13c288c592ec154ce25c534e26cb7ce0530d/src/engine/cancun.md#engine_newpayloadv3>
    pub async fn new_payload_v3(
        &self,
        payload: PayloadT::ExecutionData,
    ) -> EngineApiResult<PayloadStatus> {
        let payload_or_attrs = PayloadOrAttributes::<
            '_,
            PayloadT::ExecutionData,
            PayloadT::PayloadAttributes,
        >::from_execution_payload(&payload);
        self.inner
            .validator
            .validate_version_specific_fields(EngineApiMessageVersion::V3, payload_or_attrs)?;

        Ok(self.inner.beacon_consensus.new_payload(payload).await?)
    }

    /// Metrics version of `new_payload_v3`
    pub async fn new_payload_v3_metered(
        &self,
        payload: PayloadT::ExecutionData,
    ) -> RpcResult<PayloadStatus> {
        let start = Instant::now();

        let res = Self::new_payload_v3(self, payload).await;
        let elapsed = start.elapsed();
        self.inner.metrics.latency.new_payload_v3.record(elapsed);
        Ok(res?)
    }

    /// See also <https://github.com/ethereum/execution-apis/blob/7907424db935b93c2fe6a3c0faab943adebe8557/src/engine/prague.md#engine_newpayloadv4>
    pub async fn new_payload_v4(
        &self,
        payload: PayloadT::ExecutionData,
    ) -> EngineApiResult<PayloadStatus> {
        let payload_or_attrs = PayloadOrAttributes::<
            '_,
            PayloadT::ExecutionData,
            PayloadT::PayloadAttributes,
        >::from_execution_payload(&payload);
        self.inner
            .validator
            .validate_version_specific_fields(EngineApiMessageVersion::V4, payload_or_attrs)?;

        Ok(self.inner.beacon_consensus.new_payload(payload).await?)
    }

    /// Metrics version of `new_payload_v4`
    pub async fn new_payload_v4_metered(
        &self,
        payload: PayloadT::ExecutionData,
    ) -> RpcResult<PayloadStatus> {
        let start = Instant::now();
        let res = Self::new_payload_v4(self, payload).await;

        let elapsed = start.elapsed();
        self.inner.metrics.latency.new_payload_v4.record(elapsed);
        Ok(res?)
    }

    /// Handler for `engine_newPayloadV5`
    ///
    /// Post-Amsterdam payload handler.
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/main/src/engine/amsterdam.md#engine_newpayloadv5>
    pub async fn new_payload_v5(
        &self,
        payload: PayloadT::ExecutionData,
    ) -> EngineApiResult<PayloadStatus> {
        let payload_or_attrs = PayloadOrAttributes::<
            '_,
            PayloadT::ExecutionData,
            PayloadT::PayloadAttributes,
        >::from_execution_payload(&payload);
        self.inner
            .validator
            .validate_version_specific_fields(EngineApiMessageVersion::V5, payload_or_attrs)?;
        Ok(self.inner.beacon_consensus.new_payload(payload).await?)
    }

    /// Metrics version of `new_payload_v5`
    pub async fn new_payload_v5_metered(
        &self,
        payload: PayloadT::ExecutionData,
    ) -> RpcResult<PayloadStatus> {
        let start = Instant::now();
        let res = Self::new_payload_v5(self, payload).await;
        let elapsed = start.elapsed();
        self.inner.metrics.latency.new_payload_v5.record(elapsed);
        Ok(res?)
    }

    /// Handler for `engine_newPayloadV6`.
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/main/src/engine/bogota.md#engine_newpayloadv6>
    pub async fn new_payload_v6(
        &self,
        payload: PayloadT::ExecutionData,
    ) -> EngineApiResult<PayloadStatusV2> {
        let payload_or_attrs = PayloadOrAttributes::<
            '_,
            PayloadT::ExecutionData,
            PayloadT::PayloadAttributes,
        >::from_execution_payload(&payload);
        self.inner
            .validator
            .validate_version_specific_fields(EngineApiMessageVersion::V6, payload_or_attrs)?;

        Ok(self.inner.beacon_consensus.new_payload(payload).await?.into())
    }

    /// Metrics version of `new_payload_v6`.
    pub async fn new_payload_v6_metered(
        &self,
        payload: PayloadT::ExecutionData,
    ) -> EngineApiResult<PayloadStatusV2> {
        let start = Instant::now();
        let result = Self::new_payload_v6(self, payload).await;
        self.inner.metrics.latency.new_payload_v6.record(start.elapsed());
        result
    }

    /// Returns whether the engine accepts execution requests hash.
    pub fn accept_execution_requests_hash(&self) -> bool {
        self.inner.accept_execution_requests_hash
    }
}

impl<Provider, EngineT, Pool, Validator, ChainSpec>
    EngineApi<Provider, EngineT, Pool, Validator, ChainSpec>
where
    Provider: HeaderProvider + BlockReader + StateProviderFactory + BalProvider + 'static,
    EngineT: EngineTypes,
    Pool: TransactionPool + 'static,
    Validator: EngineApiValidator<EngineT>,
    ChainSpec: EthereumHardforks + Send + Sync + 'static,
{
    /// Sends a message to the beacon consensus engine to update the fork choice _without_
    /// withdrawals.
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/3d627c95a4d3510a8187dd02e0250ecb4331d27e/src/engine/paris.md#engine_forkchoiceUpdatedV1>
    ///
    /// Caution: This should not accept the `withdrawals` field
    pub async fn fork_choice_updated_v1(
        &self,
        state: ForkchoiceState,
        payload_attrs: Option<EngineT::PayloadAttributes>,
    ) -> EngineApiResult<ForkchoiceUpdated> {
        self.validate_and_execute_forkchoice(EngineApiMessageVersion::V1, state, payload_attrs)
            .await
    }

    /// Metrics version of `fork_choice_updated_v1`
    pub async fn fork_choice_updated_v1_metered(
        &self,
        state: ForkchoiceState,
        payload_attrs: Option<EngineT::PayloadAttributes>,
    ) -> EngineApiResult<ForkchoiceUpdated> {
        let start = Instant::now();
        let res = Self::fork_choice_updated_v1(self, state, payload_attrs).await;
        self.inner.metrics.latency.fork_choice_updated_v1.record(start.elapsed());
        res
    }

    /// Sends a message to the beacon consensus engine to update the fork choice _with_ withdrawals,
    /// but only _after_ shanghai.
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/3d627c95a4d3510a8187dd02e0250ecb4331d27e/src/engine/shanghai.md#engine_forkchoiceupdatedv2>
    pub async fn fork_choice_updated_v2(
        &self,
        state: ForkchoiceState,
        payload_attrs: Option<EngineT::PayloadAttributes>,
    ) -> EngineApiResult<ForkchoiceUpdated> {
        self.validate_and_execute_forkchoice(EngineApiMessageVersion::V2, state, payload_attrs)
            .await
    }

    /// Metrics version of `fork_choice_updated_v2`
    pub async fn fork_choice_updated_v2_metered(
        &self,
        state: ForkchoiceState,
        payload_attrs: Option<EngineT::PayloadAttributes>,
    ) -> EngineApiResult<ForkchoiceUpdated> {
        let start = Instant::now();
        let res = Self::fork_choice_updated_v2(self, state, payload_attrs).await;
        self.inner.metrics.latency.fork_choice_updated_v2.record(start.elapsed());
        res
    }

    /// Sends a message to the beacon consensus engine to update the fork choice _with_ withdrawals,
    /// but only _after_ cancun.
    ///
    /// See also  <https://github.com/ethereum/execution-apis/blob/main/src/engine/cancun.md#engine_forkchoiceupdatedv3>
    pub async fn fork_choice_updated_v3(
        &self,
        state: ForkchoiceState,
        payload_attrs: Option<EngineT::PayloadAttributes>,
    ) -> EngineApiResult<ForkchoiceUpdated> {
        self.validate_and_execute_forkchoice(EngineApiMessageVersion::V3, state, payload_attrs)
            .await
    }

    /// Metrics version of `fork_choice_updated_v3`
    pub async fn fork_choice_updated_v3_metered(
        &self,
        state: ForkchoiceState,
        payload_attrs: Option<EngineT::PayloadAttributes>,
    ) -> EngineApiResult<ForkchoiceUpdated> {
        let start = Instant::now();
        let res = Self::fork_choice_updated_v3(self, state, payload_attrs).await;
        self.inner.metrics.latency.fork_choice_updated_v3.record(start.elapsed());
        res
    }

    /// Sends a message to the beacon consensus engine to update the fork choice _with_ slot number,
    /// but only _after_ amsterdam.
    ///
    /// See also  <https://github.com/ethereum/execution-apis/blob/main/src/engine/amsterdam.md#engine_forkchoiceupdatedv4>
    pub async fn fork_choice_updated_v4(
        &self,
        state: ForkchoiceState,
        payload_attrs: Option<EngineT::PayloadAttributes>,
        custody_columns: Option<B128>,
    ) -> EngineApiResult<ForkchoiceUpdated> {
        if let Some(custody_columns) = custody_columns {
            self.inner.cell_custody.set_from_engine_api(custody_columns);
        }
        self.validate_and_execute_forkchoice(EngineApiMessageVersion::V4, state, payload_attrs)
            .await
    }

    /// Metrics version of `fork_choice_updated_v4`
    pub async fn fork_choice_updated_v4_metered(
        &self,
        state: ForkchoiceState,
        payload_attrs: Option<EngineT::PayloadAttributes>,
        custody_columns: Option<B128>,
    ) -> EngineApiResult<ForkchoiceUpdated> {
        let start = Instant::now();
        let res = Self::fork_choice_updated_v4(self, state, payload_attrs, custody_columns).await;
        self.inner.metrics.latency.fork_choice_updated_v4.record(start.elapsed());
        res
    }

    /// Handler for `engine_forkchoiceUpdatedV5`.
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/main/src/engine/bogota.md#engine_forkchoiceupdatedv5>
    pub async fn fork_choice_updated_v5(
        &self,
        state: ForkchoiceState,
        payload_attrs: Option<EngineT::PayloadAttributes>,
        custody_columns: Option<B128>,
    ) -> EngineApiResult<ForkchoiceUpdatedResponseV2> {
        if let Some(custody_columns) = custody_columns {
            self.inner.cell_custody.set_from_engine_api(custody_columns);
        }

        // Todo: Validate IL and populate `inclusion_list_satisfied` properly and test.
        Ok(self
            .validate_and_execute_forkchoice(EngineApiMessageVersion::V5, state, payload_attrs)
            .await?
            .into())
    }

    /// Metrics version of `fork_choice_updated_v5`
    pub async fn fork_choice_updated_v5_metered(
        &self,
        state: ForkchoiceState,
        payload_attrs: Option<EngineT::PayloadAttributes>,
        custody_columns: Option<B128>,
    ) -> EngineApiResult<ForkchoiceUpdatedResponseV2> {
        let start = Instant::now();
        let res = Self::fork_choice_updated_v5(self, state, payload_attrs, custody_columns).await;
        self.inner.metrics.latency.fork_choice_updated_v5.record(start.elapsed());
        res
    }

    /// Builds an EIP-7805 inclusion list from the local transaction pool.
    pub fn get_inclusion_list_v1(&self) -> EngineApiResult<Vec<Bytes>> {
        let mut total_size = 0;
        let mut inclusion_list = Vec::new();

        for pool_tx in self.inner.tx_pool.best_transactions().without_blobs().without_updates() {
            let encoded = pool_tx.encoded_2718_consensus();
            let new_size = total_size + encoded.len();
            if new_size > MAX_BYTES_PER_INCLUSION_LIST as usize {
                break
            }

            total_size = new_size;
            inclusion_list.push(encoded);
        }

        Ok(inclusion_list)
    }

    /// Metrics version of `get_inclusion_list_v1`.
    pub fn get_inclusion_list_v1_metered(&self) -> EngineApiResult<Vec<Bytes>> {
        let start = Instant::now();
        let result = Self::get_inclusion_list_v1(self);
        self.inner.metrics.latency.get_inclusion_list_v1.record(start.elapsed());
        result
    }

    /// Helper function for retrieving the build payload by id.
    async fn get_built_payload(
        &self,
        payload_id: PayloadId,
    ) -> EngineApiResult<EngineT::BuiltPayload> {
        self.inner
            .payload_store
            .resolve(payload_id)
            .await
            .ok_or(EngineApiError::UnknownPayload)?
            .map_err(|_| EngineApiError::UnknownPayload)
    }

    /// Helper function for validating the payload timestamp and retrieving & converting the payload
    /// into desired envelope.
    async fn get_payload_inner<R>(
        &self,
        payload_id: PayloadId,
        version: EngineApiMessageVersion,
    ) -> EngineApiResult<R>
    where
        EngineT::BuiltPayload: TryInto<R>,
    {
        // Validate timestamp according to engine rules
        // Enforces Osaka restrictions on `getPayloadV4`.
        let timestamp = self.get_payload_timestamp(payload_id).await?;
        validate_payload_timestamp(
            &self.inner.chain_spec,
            version,
            timestamp,
            MessageValidationKind::GetPayload,
        )?;

        // Now resolve the payload
        self.get_built_payload(payload_id).await?.try_into().map_err(|_| {
            warn!(?version, "could not transform built payload");
            EngineApiError::UnknownPayload
        })
    }

    /// Returns the most recent version of the payload that is available in the corresponding
    /// payload build process at the time of receiving this call.
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/3d627c95a4d3510a8187dd02e0250ecb4331d27e/src/engine/paris.md#engine_getPayloadV1>
    ///
    /// Caution: This should not return the `withdrawals` field
    ///
    /// Note:
    /// > Provider software MAY stop the corresponding build process after serving this call.
    pub async fn get_payload_v1(
        &self,
        payload_id: PayloadId,
    ) -> EngineApiResult<EngineT::ExecutionPayloadEnvelopeV1> {
        self.get_built_payload(payload_id).await?.try_into().map_err(|_| {
            warn!(version = ?EngineApiMessageVersion::V1, "could not transform built payload");
            EngineApiError::UnknownPayload
        })
    }

    /// Metrics version of `get_payload_v1`
    pub async fn get_payload_v1_metered(
        &self,
        payload_id: PayloadId,
    ) -> EngineApiResult<EngineT::ExecutionPayloadEnvelopeV1> {
        let start = Instant::now();
        let res = Self::get_payload_v1(self, payload_id).await;
        self.inner.metrics.latency.get_payload_v1.record(start.elapsed());
        res
    }

    /// Returns the most recent version of the payload that is available in the corresponding
    /// payload build process at the time of receiving this call.
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/3d627c95a4d3510a8187dd02e0250ecb4331d27e/src/engine/shanghai.md#engine_getpayloadv2>
    ///
    /// Note:
    /// > Provider software MAY stop the corresponding build process after serving this call.
    pub async fn get_payload_v2(
        &self,
        payload_id: PayloadId,
    ) -> EngineApiResult<EngineT::ExecutionPayloadEnvelopeV2> {
        self.get_payload_inner(payload_id, EngineApiMessageVersion::V2).await
    }

    /// Metrics version of `get_payload_v2`
    pub async fn get_payload_v2_metered(
        &self,
        payload_id: PayloadId,
    ) -> EngineApiResult<EngineT::ExecutionPayloadEnvelopeV2> {
        let start = Instant::now();
        let res = Self::get_payload_v2(self, payload_id).await;
        self.inner.metrics.latency.get_payload_v2.record(start.elapsed());
        res
    }

    /// Returns the most recent version of the payload that is available in the corresponding
    /// payload build process at the time of receiving this call.
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/fe8e13c288c592ec154ce25c534e26cb7ce0530d/src/engine/cancun.md#engine_getpayloadv3>
    ///
    /// Note:
    /// > Provider software MAY stop the corresponding build process after serving this call.
    pub async fn get_payload_v3(
        &self,
        payload_id: PayloadId,
    ) -> EngineApiResult<EngineT::ExecutionPayloadEnvelopeV3> {
        self.get_payload_inner(payload_id, EngineApiMessageVersion::V3).await
    }

    /// Metrics version of `get_payload_v3`
    pub async fn get_payload_v3_metered(
        &self,
        payload_id: PayloadId,
    ) -> EngineApiResult<EngineT::ExecutionPayloadEnvelopeV3> {
        let start = Instant::now();
        let res = Self::get_payload_v3(self, payload_id).await;
        self.inner.metrics.latency.get_payload_v3.record(start.elapsed());
        res
    }

    /// Returns the most recent version of the payload that is available in the corresponding
    /// payload build process at the time of receiving this call.
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/7907424db935b93c2fe6a3c0faab943adebe8557/src/engine/prague.md#engine_getpayloadv4>
    ///
    /// Note:
    /// > Provider software MAY stop the corresponding build process after serving this call.
    pub async fn get_payload_v4(
        &self,
        payload_id: PayloadId,
    ) -> EngineApiResult<EngineT::ExecutionPayloadEnvelopeV4> {
        self.get_payload_inner(payload_id, EngineApiMessageVersion::V4).await
    }

    /// Metrics version of `get_payload_v4`
    pub async fn get_payload_v4_metered(
        &self,
        payload_id: PayloadId,
    ) -> EngineApiResult<EngineT::ExecutionPayloadEnvelopeV4> {
        let start = Instant::now();
        let res = Self::get_payload_v4(self, payload_id).await;
        self.inner.metrics.latency.get_payload_v4.record(start.elapsed());
        res
    }

    /// Handler for `engine_getPayloadV5`
    ///
    /// Returns the most recent version of the payload that is available in the corresponding
    /// payload build process at the time of receiving this call.
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/15399c2e2f16a5f800bf3f285640357e2c245ad9/src/engine/osaka.md#engine_getpayloadv5>
    ///
    /// Note:
    /// > Provider software MAY stop the corresponding build process after serving this call.
    pub async fn get_payload_v5(
        &self,
        payload_id: PayloadId,
    ) -> EngineApiResult<EngineT::ExecutionPayloadEnvelopeV5> {
        self.get_payload_inner(payload_id, EngineApiMessageVersion::V5).await
    }

    /// Metrics version of `get_payload_v5`
    pub async fn get_payload_v5_metered(
        &self,
        payload_id: PayloadId,
    ) -> EngineApiResult<EngineT::ExecutionPayloadEnvelopeV5> {
        let start = Instant::now();
        let res = Self::get_payload_v5(self, payload_id).await;
        self.inner.metrics.latency.get_payload_v5.record(start.elapsed());
        res
    }

    /// Handler for `engine_getPayloadV6`
    ///
    /// Post-Amsterdam payload handler that includes Block Access Lists (BAL).
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/main/src/engine/amsterdam.md#engine_getpayloadv6>
    pub async fn get_payload_v6(
        &self,
        payload_id: PayloadId,
    ) -> EngineApiResult<EngineT::ExecutionPayloadEnvelopeV6> {
        self.get_payload_inner(payload_id, EngineApiMessageVersion::V6).await
    }

    /// Metrics version of `get_payload_v6`
    pub async fn get_payload_v6_metered(
        &self,
        payload_id: PayloadId,
    ) -> EngineApiResult<EngineT::ExecutionPayloadEnvelopeV6> {
        let start = Instant::now();
        let res = Self::get_payload_v6(self, payload_id).await;
        self.inner.metrics.latency.get_payload_v6.record(start.elapsed());
        res
    }

    /// Fetches all the blocks for the provided range starting at `start`, containing `count`
    /// blocks and returns the mapped payload bodies.
    pub async fn get_payload_bodies_by_range_with<F, R>(
        &self,
        start: BlockNumber,
        count: u64,
        f: F,
    ) -> EngineApiResult<Vec<Option<R>>>
    where
        F: Fn(Provider::Block) -> R + Send + 'static,
        R: Send + 'static,
    {
        let (tx, rx) = oneshot::channel();
        let inner = self.inner.clone();

        self.inner.task_spawner.spawn_blocking_task(async move {
            if count > MAX_PAYLOAD_BODIES_LIMIT {
                tx.send(Err(EngineApiError::PayloadRequestTooLarge { len: count })).ok();
                return;
            }

            if start == 0 || count == 0 {
                tx.send(Err(EngineApiError::InvalidBodiesRange { start, count })).ok();
                return;
            }

            let mut result = Vec::with_capacity(count as usize);

            // -1 so range is inclusive
            let mut end = start.saturating_add(count - 1);

            // > Client software MUST NOT return trailing null values if the request extends past the current latest known block.
            // truncate the end if it's greater than the last block
            if let Ok(best_block) = inner.provider.best_block_number()
                && end > best_block {
                    end = best_block;
                }

            // Check if the requested range starts before the earliest available block due to pruning/expiry
            let earliest_block = inner.provider.earliest_block_number().unwrap_or(0);
            for num in start..=end {
                if tx.is_closed() {
                    return;
                }

                if num < earliest_block {
                    result.push(None);
                    continue;
                }
                let block_result = inner.provider.block(BlockHashOrNumber::Number(num));
                match block_result {
                    Ok(block) => {
                        result.push(block.map(&f));
                    }
                    Err(err) => {
                        tx.send(Err(EngineApiError::Internal(Box::new(err)))).ok();
                        return;
                    }
                };
            }
            tx.send(Ok(result)).ok();
        });

        rx.await.map_err(|err| EngineApiError::Internal(Box::new(err)))?
    }

    /// Returns payload bodies and their timestamps from the same block read.
    ///
    /// SSZ transports use the timestamp to select the block's fork schema. Reading it
    /// separately by block number could pair a body with another block after a reorg.
    pub async fn get_payload_bodies_by_range_with_timestamps(
        &self,
        start: BlockNumber,
        count: u64,
        include_bal: bool,
    ) -> EngineApiResult<Vec<Option<(u64, ExecutionPayloadBodyV2)>>> {
        let bodies = self
            .get_payload_bodies_by_range_with(start, count, Self::payload_body_with_timestamp)
            .await?;
        self.attach_payload_body_bals(bodies, include_bal).await
    }

    /// Records SSZ body requests in the corresponding JSON-RPC V1/V2 latency histogram.
    pub async fn get_payload_bodies_by_range_with_timestamps_metered(
        &self,
        start: BlockNumber,
        count: u64,
        include_bal: bool,
    ) -> EngineApiResult<Vec<Option<(u64, ExecutionPayloadBodyV2)>>> {
        let start_time = Instant::now();
        let result =
            self.get_payload_bodies_by_range_with_timestamps(start, count, include_bal).await;
        let latency = &self.inner.metrics.latency;
        if include_bal {
            latency.get_payload_bodies_by_range_v2.record(start_time.elapsed());
        } else {
            latency.get_payload_bodies_by_range_v1.record(start_time.elapsed());
        }
        result
    }

    /// Returns the execution payload bodies by the range starting at `start`, containing `count`
    /// blocks.
    ///
    /// WARNING: This method is associated with the `BeaconBlocksByRange` message in the consensus
    /// layer p2p specification, meaning the input should be treated as untrusted or potentially
    /// adversarial.
    ///
    /// Implementers should take care when acting on the input to this method, specifically
    /// ensuring that the range is limited properly, and that the range boundaries are computed
    /// correctly and without panics.
    pub async fn get_payload_bodies_by_range_v1(
        &self,
        start: BlockNumber,
        count: u64,
    ) -> EngineApiResult<ExecutionPayloadBodiesV1> {
        self.get_payload_bodies_by_range_with(start, count, |block| ExecutionPayloadBodyV1 {
            transactions: block.body().encoded_2718_transactions(),
            withdrawals: block.body().withdrawals().cloned().map(Withdrawals::into_inner),
        })
        .await
    }

    /// Metrics version of `get_payload_bodies_by_range_v1`
    pub async fn get_payload_bodies_by_range_v1_metered(
        &self,
        start: BlockNumber,
        count: u64,
    ) -> EngineApiResult<ExecutionPayloadBodiesV1> {
        let start_time = Instant::now();
        let res = Self::get_payload_bodies_by_range_v1(self, start, count).await;
        self.inner.metrics.latency.get_payload_bodies_by_range_v1.record(start_time.elapsed());
        res
    }

    /// Returns the execution payload bodies by the range (V2).
    ///
    /// V2 includes the `block_access_list` field for EIP-7928 BAL support.
    pub async fn get_payload_bodies_by_range_v2(
        &self,
        start: BlockNumber,
        count: u64,
    ) -> EngineApiResult<ExecutionPayloadBodiesV2> {
        Ok(self
            .get_payload_bodies_by_range_with_timestamps(start, count, true)
            .await?
            .into_iter()
            .map(|body| body.map(|(_, body)| body))
            .collect())
    }

    /// Metrics version of `get_payload_bodies_by_range_v2`
    pub async fn get_payload_bodies_by_range_v2_metered(
        &self,
        start: BlockNumber,
        count: u64,
    ) -> EngineApiResult<ExecutionPayloadBodiesV2> {
        let start_time = Instant::now();
        let res = Self::get_payload_bodies_by_range_v2(self, start, count).await;
        self.inner.metrics.latency.get_payload_bodies_by_range_v2.record(start_time.elapsed());
        res
    }

    /// Called to retrieve execution payload bodies by hashes.
    pub async fn get_payload_bodies_by_hash_with<F, R>(
        &self,
        hashes: Vec<BlockHash>,
        f: F,
    ) -> EngineApiResult<Vec<Option<R>>>
    where
        F: Fn(Provider::Block) -> R + Send + 'static,
        R: Send + 'static,
    {
        let len = hashes.len() as u64;
        if len > MAX_PAYLOAD_BODIES_LIMIT {
            return Err(EngineApiError::PayloadRequestTooLarge { len });
        }

        let (tx, rx) = oneshot::channel();
        let inner = self.inner.clone();

        self.inner.task_spawner.spawn_blocking_task(async move {
            let mut result = Vec::with_capacity(hashes.len());
            for hash in hashes {
                if tx.is_closed() {
                    return;
                }

                let block_result = inner.provider.block(BlockHashOrNumber::Hash(hash));
                match block_result {
                    Ok(block) => {
                        result.push(block.map(&f));
                    }
                    Err(err) => {
                        let _ = tx.send(Err(EngineApiError::Internal(Box::new(err))));
                        return;
                    }
                }
            }
            tx.send(Ok(result)).ok();
        });

        rx.await.map_err(|err| EngineApiError::Internal(Box::new(err)))?
    }

    /// Returns payload bodies and their timestamps from the same block read.
    pub async fn get_payload_bodies_by_hash_with_timestamps(
        &self,
        hashes: Vec<BlockHash>,
        include_bal: bool,
    ) -> EngineApiResult<Vec<Option<(u64, ExecutionPayloadBodyV2)>>> {
        let bodies =
            self.get_payload_bodies_by_hash_with(hashes, Self::payload_body_with_timestamp).await?;
        self.attach_payload_body_bals(bodies, include_bal).await
    }

    /// Records SSZ body requests in the corresponding JSON-RPC V1/V2 latency histogram.
    pub async fn get_payload_bodies_by_hash_with_timestamps_metered(
        &self,
        hashes: Vec<BlockHash>,
        include_bal: bool,
    ) -> EngineApiResult<Vec<Option<(u64, ExecutionPayloadBodyV2)>>> {
        let start = Instant::now();
        let result = self.get_payload_bodies_by_hash_with_timestamps(hashes, include_bal).await;
        let latency = &self.inner.metrics.latency;
        if include_bal {
            latency.get_payload_bodies_by_hash_v2.record(start.elapsed());
        } else {
            latency.get_payload_bodies_by_hash_v1.record(start.elapsed());
        }
        result
    }

    fn payload_body_with_timestamp(
        block: Provider::Block,
    ) -> (BlockHash, u64, ExecutionPayloadBodyV2) {
        (
            block.header().hash_slow(),
            block.header().timestamp(),
            ExecutionPayloadBodyV2 {
                transactions: block.body().encoded_2718_transactions(),
                withdrawals: block.body().withdrawals().cloned().map(Withdrawals::into_inner),
                block_access_list: None,
            },
        )
    }

    async fn attach_payload_body_bals(
        &self,
        mut bodies: Vec<Option<(BlockHash, u64, ExecutionPayloadBodyV2)>>,
        include_bal: bool,
    ) -> EngineApiResult<Vec<Option<(u64, ExecutionPayloadBodyV2)>>> {
        if include_bal {
            let hashes = bodies.iter().flatten().map(|(hash, _, _)| *hash).collect();
            let bals = self.get_block_access_lists_by_hashes(hashes).await?;
            for ((_, _, body), bal) in bodies.iter_mut().flatten().zip(bals) {
                body.block_access_list = bal;
            }
        }
        Ok(bodies
            .into_iter()
            .map(|body| body.map(|(_, timestamp, body)| (timestamp, body)))
            .collect())
    }

    async fn get_block_access_lists_by_hashes(
        &self,
        hashes: Vec<BlockHash>,
    ) -> EngineApiResult<Vec<Option<Bytes>>> {
        let len = hashes.len() as u64;
        if len > MAX_PAYLOAD_BODIES_LIMIT {
            return Err(EngineApiError::PayloadRequestTooLarge { len });
        }

        let (tx, rx) = oneshot::channel();
        let inner = self.inner.clone();

        self.inner.task_spawner.spawn_blocking_task(async move {
            if tx.is_closed() {
                return;
            }

            tx.send(
                inner
                    .provider
                    .get_bals_by_hashes(&hashes)
                    .map_err(|err| EngineApiError::Internal(Box::new(err))),
            )
            .ok();
        });

        rx.await.map_err(|err| EngineApiError::Internal(Box::new(err)))?
    }

    /// Called to retrieve execution payload bodies by hashes.
    pub async fn get_payload_bodies_by_hash_v1(
        &self,
        hashes: Vec<BlockHash>,
    ) -> EngineApiResult<ExecutionPayloadBodiesV1> {
        self.get_payload_bodies_by_hash_with(hashes, |block| ExecutionPayloadBodyV1 {
            transactions: block.body().encoded_2718_transactions(),
            withdrawals: block.body().withdrawals().cloned().map(Withdrawals::into_inner),
        })
        .await
    }

    /// Metrics version of `get_payload_bodies_by_hash_v1`
    pub async fn get_payload_bodies_by_hash_v1_metered(
        &self,
        hashes: Vec<BlockHash>,
    ) -> EngineApiResult<ExecutionPayloadBodiesV1> {
        let start = Instant::now();
        let res = Self::get_payload_bodies_by_hash_v1(self, hashes).await;
        self.inner.metrics.latency.get_payload_bodies_by_hash_v1.record(start.elapsed());
        res
    }

    /// Called to retrieve execution payload bodies by hashes (V2).
    ///
    /// V2 includes the `block_access_list` field for EIP-7928 BAL support.
    pub async fn get_payload_bodies_by_hash_v2(
        &self,
        hashes: Vec<BlockHash>,
    ) -> EngineApiResult<ExecutionPayloadBodiesV2> {
        let payload_bodies =
            self.get_payload_bodies_by_hash_with(hashes.clone(), |block| ExecutionPayloadBodyV2 {
                transactions: block.body().encoded_2718_transactions(),
                withdrawals: block.body().withdrawals().cloned().map(Withdrawals::into_inner),
                block_access_list: None,
            });
        let block_access_lists = self.get_block_access_lists_by_hashes(hashes);
        let (mut payload_bodies, block_access_lists) =
            tokio::try_join!(payload_bodies, block_access_lists)?;

        for (payload_body, block_access_list) in payload_bodies.iter_mut().zip(block_access_lists) {
            if let Some(payload_body) = payload_body {
                payload_body.block_access_list = block_access_list;
            }
        }

        Ok(payload_bodies)
    }

    /// Metrics version of `get_payload_bodies_by_hash_v2`
    pub async fn get_payload_bodies_by_hash_v2_metered(
        &self,
        hashes: Vec<BlockHash>,
    ) -> EngineApiResult<ExecutionPayloadBodiesV2> {
        let start = Instant::now();
        let res = Self::get_payload_bodies_by_hash_v2(self, hashes).await;
        self.inner.metrics.latency.get_payload_bodies_by_hash_v2.record(start.elapsed());
        res
    }

    /// Validates the `engine_forkchoiceUpdated` payload attributes and executes the forkchoice
    /// update.
    ///
    /// The payload attributes will be validated according to the engine API rules for the given
    /// message version:
    /// * If the version is [`EngineApiMessageVersion::V1`], then the payload attributes will be
    ///   validated according to the Paris rules.
    /// * If the version is [`EngineApiMessageVersion::V2`], then the payload attributes will be
    ///   validated according to the Shanghai rules, as well as the validity changes from cancun:
    ///   <https://github.com/ethereum/execution-apis/blob/584905270d8ad665718058060267061ecfd79ca5/src/engine/cancun.md#update-the-methods-of-previous-forks>
    ///
    /// * If the version above [`EngineApiMessageVersion::V3`], then the payload attributes will be
    ///   validated according to the Cancun rules.
    async fn validate_and_execute_forkchoice(
        &self,
        version: EngineApiMessageVersion,
        state: ForkchoiceState,
        payload_attrs: Option<EngineT::PayloadAttributes>,
    ) -> EngineApiResult<ForkchoiceUpdated> {
        if let Some(ref attrs) = payload_attrs {
            let attr_validation_res =
                self.inner.validator.ensure_well_formed_attributes(version, attrs);

            // From the engine API spec:
            //
            // Client software MUST ensure that payloadAttributes.timestamp is greater than
            // timestamp of a block referenced by forkchoiceState.headBlockHash. If this condition
            // isn't held client software MUST respond with -38003: Invalid payload attributes and
            // MUST NOT begin a payload build process. In such an event, the forkchoiceState
            // update MUST NOT be rolled back.
            //
            // NOTE: This also applies to cancun/shanghai-specific payload attributes.
            if let Err(err) = attr_validation_res {
                let fcu_res = self.inner.beacon_consensus.fork_choice_updated(state, None).await?;
                if fcu_res.is_invalid() || fcu_res.payload_status.is_syncing() {
                    return Ok(fcu_res)
                }
                return Err(err.into())
            }
        }

        Ok(self.inner.beacon_consensus.fork_choice_updated(state, payload_attrs).await?)
    }

    /// Returns reference to supported capabilities.
    pub fn capabilities(&self) -> &EngineCapabilities {
        &self.inner.capabilities
    }

    fn has_blobs(&self, versioned_hashes: Vec<B256>) -> EngineApiResult<Vec<bool>> {
        if versioned_hashes.len() > MAX_BLOB_LIMIT {
            return Err(EngineApiError::BlobRequestTooLarge { len: versioned_hashes.len() })
        }

        self.inner
            .tx_pool
            .has_blobs_for_versioned_hashes(&versioned_hashes)
            .map_err(|err| EngineApiError::Internal(Box::new(err)))
    }

    /// Metered version of `has_blobs`.
    pub fn has_blobs_metered(&self, versioned_hashes: Vec<B256>) -> EngineApiResult<Vec<bool>> {
        let start = Instant::now();
        let res = Self::has_blobs(self, versioned_hashes);
        self.inner.metrics.latency.has_blobs.record(start.elapsed());
        res
    }

    fn get_blobs_v1(
        &self,
        versioned_hashes: Vec<B256>,
    ) -> EngineApiResult<Vec<Option<BlobAndProofV1>>> {
        // Only allow this method before Osaka fork
        let current_timestamp =
            SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap_or_default().as_secs();
        if self.inner.chain_spec.is_osaka_active_at_timestamp(current_timestamp) {
            return Err(EngineApiError::EngineObjectValidationError(
                reth_payload_primitives::EngineObjectValidationError::UnsupportedFork,
            ));
        }

        if versioned_hashes.len() > MAX_BLOB_LIMIT {
            return Err(EngineApiError::BlobRequestTooLarge { len: versioned_hashes.len() })
        }

        self.inner
            .tx_pool
            .get_blobs_for_versioned_hashes_v1(&versioned_hashes)
            .map_err(|err| EngineApiError::Internal(Box::new(err)))
    }

    /// Metered version of `get_blobs_v1`.
    pub fn get_blobs_v1_metered(
        &self,
        versioned_hashes: Vec<B256>,
    ) -> EngineApiResult<Vec<Option<BlobAndProofV1>>> {
        let hashes_len = versioned_hashes.len();
        let start = Instant::now();
        let res = Self::get_blobs_v1(self, versioned_hashes);
        self.inner.metrics.latency.get_blobs_v1.record(start.elapsed());

        if let Ok(blobs) = &res {
            let blobs_found = blobs.iter().flatten().count();
            let blobs_missed = hashes_len - blobs_found;

            self.inner.metrics.blob_metrics.blob_count.increment(blobs_found as u64);
            self.inner.metrics.blob_metrics.blob_misses.increment(blobs_missed as u64);
        }

        res
    }

    fn get_blobs_v2(
        &self,
        versioned_hashes: Vec<B256>,
    ) -> EngineApiResult<Option<Vec<BlobAndProofV2>>> {
        // Check if Osaka fork is active
        let current_timestamp =
            SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap_or_default().as_secs();
        if !self.inner.chain_spec.is_osaka_active_at_timestamp(current_timestamp) {
            return Err(EngineApiError::EngineObjectValidationError(
                reth_payload_primitives::EngineObjectValidationError::UnsupportedFork,
            ));
        }

        if versioned_hashes.len() > MAX_BLOB_LIMIT {
            return Err(EngineApiError::BlobRequestTooLarge { len: versioned_hashes.len() })
        }

        self.inner
            .tx_pool
            .get_blobs_for_versioned_hashes_v2(&versioned_hashes)
            .map_err(|err| EngineApiError::Internal(Box::new(err)))
    }

    fn get_blobs_v3(
        &self,
        versioned_hashes: Vec<B256>,
    ) -> EngineApiResult<Option<Vec<Option<BlobAndProofV2>>>> {
        // Check if Osaka fork is active
        let current_timestamp =
            SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap_or_default().as_secs();
        if !self.inner.chain_spec.is_osaka_active_at_timestamp(current_timestamp) {
            return Err(EngineApiError::EngineObjectValidationError(
                reth_payload_primitives::EngineObjectValidationError::UnsupportedFork,
            ));
        }

        if versioned_hashes.len() > MAX_BLOB_LIMIT {
            return Err(EngineApiError::BlobRequestTooLarge { len: versioned_hashes.len() })
        }

        // Spec requires returning `null` if syncing.
        if (*self.inner.is_syncing)() {
            return Ok(None)
        }

        self.inner
            .tx_pool
            .get_blobs_for_versioned_hashes_v3(&versioned_hashes)
            .map(Some)
            .map_err(|err| EngineApiError::Internal(Box::new(err)))
    }

    fn get_blobs_v4(
        &self,
        versioned_hashes: Vec<B256>,
        indices_bitarray: B128,
    ) -> EngineApiResult<Option<Vec<Option<BlobCellsAndProofsV1>>>> {
        // Engine API bitvectors are little-endian, while B128 integer conversions are
        // big-endian. The transaction pool uses the latter representation as a numeric cell mask.
        let indices_bitarray = B128::from(u128::from_le_bytes(indices_bitarray.into()));
        let current_timestamp =
            SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap_or_default().as_secs();
        if !self.inner.chain_spec.is_osaka_active_at_timestamp(current_timestamp) {
            return Err(EngineApiError::EngineObjectValidationError(
                reth_payload_primitives::EngineObjectValidationError::UnsupportedFork,
            ));
        }

        if versioned_hashes.len() > MAX_BLOB_LIMIT {
            return Err(EngineApiError::BlobRequestTooLarge { len: versioned_hashes.len() })
        }

        // Spec requires returning `null` if syncing.
        if (*self.inner.is_syncing)() {
            return Ok(None)
        }

        self.inner
            .tx_pool
            .get_blobs_for_versioned_hashes_v4(&versioned_hashes, indices_bitarray)
            .map(Some)
            .map_err(|err| EngineApiError::Internal(Box::new(err)))
    }

    /// Metered version of `get_blobs_v2`.
    pub fn get_blobs_v2_metered(
        &self,
        versioned_hashes: Vec<B256>,
    ) -> EngineApiResult<Option<Vec<BlobAndProofV2>>> {
        let hashes_len = versioned_hashes.len();
        let start = Instant::now();
        let res = Self::get_blobs_v2(self, versioned_hashes);
        self.inner.metrics.latency.get_blobs_v2.record(start.elapsed());

        if let Ok(blobs) = &res {
            let blobs_found = blobs.iter().flatten().count();

            self.inner
                .metrics
                .blob_metrics
                .get_blobs_requests_blobs_total
                .increment(hashes_len as u64);
            self.inner
                .metrics
                .blob_metrics
                .get_blobs_requests_blobs_in_blobpool_total
                .increment(blobs_found as u64);

            if blobs_found == hashes_len {
                self.inner.metrics.blob_metrics.get_blobs_requests_success_total.increment(1);
            } else {
                self.inner.metrics.blob_metrics.get_blobs_requests_failure_total.increment(1);
            }
        } else {
            self.inner.metrics.blob_metrics.get_blobs_requests_failure_total.increment(1);
        }

        res
    }

    /// Metered version of `get_blobs_v3`.
    pub fn get_blobs_v3_metered(
        &self,
        versioned_hashes: Vec<B256>,
    ) -> EngineApiResult<Option<Vec<Option<BlobAndProofV2>>>> {
        let hashes_len = versioned_hashes.len();
        let start = Instant::now();
        let res = Self::get_blobs_v3(self, versioned_hashes);
        self.inner.metrics.latency.get_blobs_v3.record(start.elapsed());

        if let Ok(Some(blobs)) = &res {
            let blobs_found = blobs.iter().flatten().count();
            let blobs_missed = hashes_len - blobs_found;

            self.inner.metrics.blob_metrics.blob_count.increment(blobs_found as u64);
            self.inner.metrics.blob_metrics.blob_misses.increment(blobs_missed as u64);
        }

        res
    }

    /// Metered version of `get_blobs_v4`.
    pub fn get_blobs_v4_metered(
        &self,
        versioned_hashes: Vec<B256>,
        indices_bitarray: B128,
    ) -> EngineApiResult<Option<Vec<Option<BlobCellsAndProofsV1>>>> {
        let hashes_len = versioned_hashes.len();
        let start = Instant::now();
        let res = Self::get_blobs_v4(self, versioned_hashes, indices_bitarray);
        self.inner.metrics.latency.get_blobs_v4.record(start.elapsed());

        if let Ok(Some(blobs)) = &res {
            let blobs_found = blobs.iter().flatten().count();
            let blobs_missed = hashes_len - blobs_found;

            self.inner.metrics.blob_metrics.blob_count.increment(blobs_found as u64);
            self.inner.metrics.blob_metrics.blob_misses.increment(blobs_missed as u64);
        }

        res
    }
}

// This is the concrete ethereum engine API implementation.
#[async_trait]
impl<Provider, EngineT, Pool, Validator, ChainSpec> EngineApiServer<EngineT>
    for EngineApi<Provider, EngineT, Pool, Validator, ChainSpec>
where
    Provider: HeaderProvider + BlockReader + StateProviderFactory + BalProvider + 'static,
    EngineT: EngineTypes<ExecutionData = ExecutionData>,
    Pool: TransactionPool + 'static,
    Validator: EngineApiValidator<EngineT>,
    ChainSpec: EthereumHardforks + Send + Sync + 'static,
{
    /// Handler for `engine_newPayloadV1`
    /// See also <https://github.com/ethereum/execution-apis/blob/3d627c95a4d3510a8187dd02e0250ecb4331d27e/src/engine/paris.md#engine_newpayloadv1>
    /// Caution: This should not accept the `withdrawals` field
    async fn new_payload_v1(&self, payload: ExecutionPayloadV1) -> RpcResult<PayloadStatus> {
        trace!(target: "rpc::engine", "Serving engine_newPayloadV1");
        let payload =
            ExecutionData { payload: payload.into(), sidecar: ExecutionPayloadSidecar::none() };
        Ok(self.new_payload_v1_metered(payload).await?)
    }

    /// Handler for `engine_newPayloadV2`
    /// See also <https://github.com/ethereum/execution-apis/blob/584905270d8ad665718058060267061ecfd79ca5/src/engine/shanghai.md#engine_newpayloadv2>
    async fn new_payload_v2(&self, payload: ExecutionPayloadInputV2) -> RpcResult<PayloadStatus> {
        trace!(target: "rpc::engine", "Serving engine_newPayloadV2");
        let payload = ExecutionData {
            payload: payload.into_payload(),
            sidecar: ExecutionPayloadSidecar::none(),
        };

        Ok(self.new_payload_v2_metered(payload).await?)
    }

    /// Handler for `engine_newPayloadV3`
    /// See also <https://github.com/ethereum/execution-apis/blob/fe8e13c288c592ec154ce25c534e26cb7ce0530d/src/engine/cancun.md#engine_newpayloadv3>
    async fn new_payload_v3(
        &self,
        payload: ExecutionPayloadV3,
        versioned_hashes: Vec<B256>,
        parent_beacon_block_root: B256,
    ) -> RpcResult<PayloadStatus> {
        trace!(target: "rpc::engine", "Serving engine_newPayloadV3");
        let payload = ExecutionData {
            payload: payload.into(),
            sidecar: ExecutionPayloadSidecar::v3(CancunPayloadFields {
                versioned_hashes,
                parent_beacon_block_root,
            }),
        };

        Ok(self.new_payload_v3_metered(payload).await?)
    }

    /// Handler for `engine_newPayloadV4`
    /// See also <https://github.com/ethereum/execution-apis/blob/03911ffc053b8b806123f1fc237184b0092a485a/src/engine/prague.md#engine_newpayloadv4>
    async fn new_payload_v4(
        &self,
        payload: ExecutionPayloadV3,
        versioned_hashes: Vec<B256>,
        parent_beacon_block_root: B256,
        requests: RequestsOrHash,
    ) -> RpcResult<PayloadStatus> {
        trace!(target: "rpc::engine", "Serving engine_newPayloadV4");

        // Accept requests as a hash only if it is explicitly allowed
        if requests.is_hash() && !self.inner.accept_execution_requests_hash {
            return Err(EngineApiError::UnexpectedRequestsHash.into());
        }

        let payload = ExecutionData {
            payload: payload.into(),
            sidecar: ExecutionPayloadSidecar::v4(
                CancunPayloadFields { versioned_hashes, parent_beacon_block_root },
                PraguePayloadFields { requests },
            ),
        };

        Ok(self.new_payload_v4_metered(payload).await?)
    }

    /// Handler for `engine_newPayloadV5`
    ///
    /// Post Amsterdam payload handler.
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/main/src/engine/amsterdam.md#engine_newpayloadv5>
    async fn new_payload_v5(
        &self,
        payload: ExecutionPayloadV4,
        versioned_hashes: Vec<B256>,
        parent_beacon_block_root: B256,
        requests: RequestsOrHash,
    ) -> RpcResult<PayloadStatus> {
        trace!(target: "rpc::engine", "Serving engine_newPayloadV5");
        // Accept requests as a hash only if it is explicitly allowed.
        if requests.is_hash() && !self.inner.accept_execution_requests_hash {
            return Err(EngineApiError::UnexpectedRequestsHash.into());
        }

        let payload = ExecutionData {
            payload: payload.into(),
            sidecar: ExecutionPayloadSidecar::v4(
                CancunPayloadFields { versioned_hashes, parent_beacon_block_root },
                PraguePayloadFields { requests },
            ),
        };

        Ok(self.new_payload_v5_metered(payload).await?)
    }

    /// Handler for `engine_newPayloadV6`.
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/main/src/engine/bogota.md#engine_newpayloadv6>
    async fn new_payload_v6(
        &self,
        payload: ExecutionPayloadV4,
        versioned_hashes: Vec<B256>,
        parent_beacon_block_root: B256,
        execution_requests: RequestsOrHash,
        inclusion_list_transactions: Vec<Bytes>,
    ) -> RpcResult<PayloadStatusV2> {
        trace!(target: "rpc::engine", "Serving engine_newPayloadV6");
        if execution_requests.is_hash() && !self.inner.accept_execution_requests_hash {
            return Err(EngineApiError::UnexpectedRequestsHash.into());
        }

        let payload = ExecutionData {
            payload: payload.into(),
            sidecar: ExecutionPayloadSidecar::v6(
                CancunPayloadFields { versioned_hashes, parent_beacon_block_root },
                PraguePayloadFields { requests: execution_requests },
                BogotaPayloadFields { inclusion_list_transactions },
            ),
        };

        // TODO: perform structural validation of the inclusion list transactions and populate
        // `inclusion_list_satisfied` for VALID payloads
        Ok(self.new_payload_v6_metered(payload).await?)
    }

    /// Handler for `engine_forkchoiceUpdatedV1`
    /// See also <https://github.com/ethereum/execution-apis/blob/3d627c95a4d3510a8187dd02e0250ecb4331d27e/src/engine/paris.md#engine_forkchoiceupdatedv1>
    ///
    /// Caution: This should not accept the `withdrawals` field
    async fn fork_choice_updated_v1(
        &self,
        fork_choice_state: ForkchoiceState,
        payload_attributes: Option<EngineT::PayloadAttributes>,
    ) -> RpcResult<ForkchoiceUpdated> {
        trace!(target: "rpc::engine", "Serving engine_forkchoiceUpdatedV1");
        Ok(self.fork_choice_updated_v1_metered(fork_choice_state, payload_attributes).await?)
    }

    /// Handler for `engine_forkchoiceUpdatedV2`
    /// See also <https://github.com/ethereum/execution-apis/blob/3d627c95a4d3510a8187dd02e0250ecb4331d27e/src/engine/shanghai.md#engine_forkchoiceupdatedv2>
    async fn fork_choice_updated_v2(
        &self,
        fork_choice_state: ForkchoiceState,
        payload_attributes: Option<EngineT::PayloadAttributes>,
    ) -> RpcResult<ForkchoiceUpdated> {
        trace!(target: "rpc::engine", "Serving engine_forkchoiceUpdatedV2");
        Ok(self.fork_choice_updated_v2_metered(fork_choice_state, payload_attributes).await?)
    }

    /// Handler for `engine_forkchoiceUpdatedV3`
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/main/src/engine/cancun.md#engine_forkchoiceupdatedv3>
    async fn fork_choice_updated_v3(
        &self,
        fork_choice_state: ForkchoiceState,
        payload_attributes: Option<EngineT::PayloadAttributes>,
    ) -> RpcResult<ForkchoiceUpdated> {
        trace!(target: "rpc::engine", "Serving engine_forkchoiceUpdatedV3");
        Ok(self.fork_choice_updated_v3_metered(fork_choice_state, payload_attributes).await?)
    }

    /// Handler for `engine_forkchoiceUpdatedV4`
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/main/src/engine/amsterdam.md#engine_forkchoiceupdatedv4>
    async fn fork_choice_updated_v4(
        &self,
        fork_choice_state: ForkchoiceState,
        payload_attributes: Option<EngineT::PayloadAttributes>,
        custody_columns: Option<B128>,
    ) -> RpcResult<ForkchoiceUpdated> {
        trace!(target: "rpc::engine", "Serving engine_forkchoiceUpdatedV4");
        Ok(self
            .fork_choice_updated_v4_metered(fork_choice_state, payload_attributes, custody_columns)
            .await?)
    }

    /// Handler for `engine_forkchoiceUpdatedV5`.
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/main/src/engine/bogota.md#engine_forkchoiceupdatedv5>
    async fn fork_choice_updated_v5(
        &self,
        fork_choice_state: ForkchoiceState,
        payload_attributes: Option<EngineT::PayloadAttributes>,
        custody_columns: Option<B128>,
    ) -> RpcResult<ForkchoiceUpdatedResponseV2> {
        trace!(target: "rpc::engine", "Serving engine_forkchoiceUpdatedV5");
        Ok(self
            .fork_choice_updated_v5_metered(fork_choice_state, payload_attributes, custody_columns)
            .await?)
    }

    /// Handler for `engine_getPayloadV1`
    ///
    /// Returns the most recent version of the payload that is available in the corresponding
    /// payload build process at the time of receiving this call.
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/3d627c95a4d3510a8187dd02e0250ecb4331d27e/src/engine/paris.md#engine_getPayloadV1>
    ///
    /// Caution: This should not return the `withdrawals` field
    ///
    /// Note:
    /// > Provider software MAY stop the corresponding build process after serving this call.
    async fn get_payload_v1(
        &self,
        payload_id: PayloadId,
    ) -> RpcResult<EngineT::ExecutionPayloadEnvelopeV1> {
        trace!(target: "rpc::engine", "Serving engine_getPayloadV1");
        Ok(self.get_payload_v1_metered(payload_id).await?)
    }

    /// Handler for `engine_getPayloadV2`
    ///
    /// Returns the most recent version of the payload that is available in the corresponding
    /// payload build process at the time of receiving this call.
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/3d627c95a4d3510a8187dd02e0250ecb4331d27e/src/engine/shanghai.md#engine_getpayloadv2>
    ///
    /// Note:
    /// > Provider software MAY stop the corresponding build process after serving this call.
    async fn get_payload_v2(
        &self,
        payload_id: PayloadId,
    ) -> RpcResult<EngineT::ExecutionPayloadEnvelopeV2> {
        debug!(target: "rpc::engine", id = %payload_id, "Serving engine_getPayloadV2");
        Ok(self.get_payload_v2_metered(payload_id).await?)
    }

    /// Handler for `engine_getPayloadV3`
    ///
    /// Returns the most recent version of the payload that is available in the corresponding
    /// payload build process at the time of receiving this call.
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/fe8e13c288c592ec154ce25c534e26cb7ce0530d/src/engine/cancun.md#engine_getpayloadv3>
    ///
    /// Note:
    /// > Provider software MAY stop the corresponding build process after serving this call.
    async fn get_payload_v3(
        &self,
        payload_id: PayloadId,
    ) -> RpcResult<EngineT::ExecutionPayloadEnvelopeV3> {
        trace!(target: "rpc::engine", "Serving engine_getPayloadV3");
        Ok(self.get_payload_v3_metered(payload_id).await?)
    }

    /// Handler for `engine_getPayloadV4`
    ///
    /// Returns the most recent version of the payload that is available in the corresponding
    /// payload build process at the time of receiving this call.
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/main/src/engine/prague.md#engine_getpayloadv4>
    ///
    /// Note:
    /// > Provider software MAY stop the corresponding build process after serving this call.
    async fn get_payload_v4(
        &self,
        payload_id: PayloadId,
    ) -> RpcResult<EngineT::ExecutionPayloadEnvelopeV4> {
        trace!(target: "rpc::engine", "Serving engine_getPayloadV4");
        Ok(self.get_payload_v4_metered(payload_id).await?)
    }

    /// Handler for `engine_getPayloadV5`
    ///
    /// Returns the most recent version of the payload that is available in the corresponding
    /// payload build process at the time of receiving this call.
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/15399c2e2f16a5f800bf3f285640357e2c245ad9/src/engine/osaka.md#engine_getpayloadv5>
    ///
    /// Note:
    /// > Provider software MAY stop the corresponding build process after serving this call.
    async fn get_payload_v5(
        &self,
        payload_id: PayloadId,
    ) -> RpcResult<EngineT::ExecutionPayloadEnvelopeV5> {
        trace!(target: "rpc::engine", "Serving engine_getPayloadV5");
        Ok(self.get_payload_v5_metered(payload_id).await?)
    }

    /// Handler for `engine_getPayloadV6`
    ///
    /// Post-Amsterdam payload handler that includes Block Access Lists (BAL).
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/main/src/engine/amsterdam.md#engine_getpayloadv6>
    async fn get_payload_v6(
        &self,
        payload_id: PayloadId,
    ) -> RpcResult<EngineT::ExecutionPayloadEnvelopeV6> {
        trace!(target: "rpc::engine", "Serving engine_getPayloadV6");
        Ok(self.get_payload_v6_metered(payload_id).await?)
    }

    /// Handler for `engine_getInclusionListV1`.
    ///
    /// See also <https://github.com/ethereum/execution-apis/pull/609>.
    async fn get_inclusion_list_v1(&self) -> RpcResult<Vec<Bytes>> {
        trace!(target: "rpc::engine", "Serving engine_getInclusionListV1");
        Ok(self.get_inclusion_list_v1_metered()?)
    }

    /// Handler for `engine_getPayloadBodiesByHashV1`
    /// See also <https://github.com/ethereum/execution-apis/blob/6452a6b194d7db269bf1dbd087a267251d3cc7f8/src/engine/shanghai.md#engine_getpayloadbodiesbyhashv1>
    async fn get_payload_bodies_by_hash_v1(
        &self,
        block_hashes: Vec<BlockHash>,
    ) -> RpcResult<ExecutionPayloadBodiesV1> {
        trace!(target: "rpc::engine", "Serving engine_getPayloadBodiesByHashV1");
        Ok(self.get_payload_bodies_by_hash_v1_metered(block_hashes).await?)
    }

    /// Handler for `engine_getPayloadBodiesByHashV2`
    ///
    /// V2 includes the `block_access_list` field for EIP-7928 BAL support.
    ///
    /// See also <https://eips.ethereum.org/EIPS/eip-7928>
    async fn get_payload_bodies_by_hash_v2(
        &self,
        block_hashes: Vec<BlockHash>,
    ) -> RpcResult<ExecutionPayloadBodiesV2> {
        trace!(target: "rpc::engine", "Serving engine_getPayloadBodiesByHashV2");
        Ok(self.get_payload_bodies_by_hash_v2_metered(block_hashes).await?)
    }

    /// Handler for `engine_getPayloadBodiesByRangeV1`
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/6452a6b194d7db269bf1dbd087a267251d3cc7f8/src/engine/shanghai.md#engine_getpayloadbodiesbyrangev1>
    ///
    /// Returns the execution payload bodies by the range starting at `start`, containing `count`
    /// blocks.
    ///
    /// WARNING: This method is associated with the `BeaconBlocksByRange` message in the consensus
    /// layer p2p specification, meaning the input should be treated as untrusted or potentially
    /// adversarial.
    ///
    /// Implementers should take care when acting on the input to this method, specifically
    /// ensuring that the range is limited properly, and that the range boundaries are computed
    /// correctly and without panics.
    ///
    /// Note: If a block is pre shanghai, `withdrawals` field will be `null`.
    async fn get_payload_bodies_by_range_v1(
        &self,
        start: U64,
        count: U64,
    ) -> RpcResult<ExecutionPayloadBodiesV1> {
        trace!(target: "rpc::engine", "Serving engine_getPayloadBodiesByRangeV1");
        Ok(self.get_payload_bodies_by_range_v1_metered(start.to(), count.to()).await?)
    }

    /// Handler for `engine_getPayloadBodiesByRangeV2`
    ///
    /// V2 includes the `block_access_list` field for EIP-7928 BAL support.
    ///
    /// See also <https://eips.ethereum.org/EIPS/eip-7928>
    async fn get_payload_bodies_by_range_v2(
        &self,
        start: U64,
        count: U64,
    ) -> RpcResult<ExecutionPayloadBodiesV2> {
        trace!(target: "rpc::engine", "Serving engine_getPayloadBodiesByRangeV2");
        Ok(self.get_payload_bodies_by_range_v2_metered(start.to(), count.to()).await?)
    }

    /// Handler for `engine_getClientVersionV1`
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/03911ffc053b8b806123f1fc237184b0092a485a/src/engine/identification.md>
    async fn get_client_version_v1(
        &self,
        client: ClientVersionV1,
    ) -> RpcResult<Vec<ClientVersionV1>> {
        trace!(target: "rpc::engine", "Serving engine_getClientVersionV1");
        Ok(Self::get_client_version_v1(self, client)?)
    }

    /// Handler for `engine_exchangeCapabilitiesV1`
    /// See also <https://github.com/ethereum/execution-apis/blob/6452a6b194d7db269bf1dbd087a267251d3cc7f8/src/engine/common.md#capabilities>
    async fn exchange_capabilities(&self, capabilities: Vec<String>) -> RpcResult<Vec<String>> {
        trace!(target: "rpc::engine", "Serving engine_exchangeCapabilities");

        let el_caps = self.capabilities();
        el_caps.log_capability_mismatches(&capabilities);

        Ok(el_caps.list())
    }

    async fn has_blobs(&self, versioned_hashes: Vec<B256>) -> RpcResult<Vec<bool>> {
        trace!(target: "rpc::engine", "Serving engine_hasBlobs");
        Ok(self.has_blobs_metered(versioned_hashes)?)
    }

    async fn get_blobs_v1(
        &self,
        versioned_hashes: Vec<B256>,
    ) -> RpcResult<Vec<Option<BlobAndProofV1>>> {
        trace!(target: "rpc::engine", "Serving engine_getBlobsV1");
        Ok(self.get_blobs_v1_metered(versioned_hashes)?)
    }

    async fn get_blobs_v2(
        &self,
        versioned_hashes: Vec<B256>,
    ) -> RpcResult<Option<Vec<BlobAndProofV2>>> {
        trace!(target: "rpc::engine", "Serving engine_getBlobsV2");
        Ok(self.get_blobs_v2_metered(versioned_hashes)?)
    }

    async fn get_blobs_v3(
        &self,
        versioned_hashes: Vec<B256>,
    ) -> RpcResult<Option<Vec<Option<BlobAndProofV2>>>> {
        trace!(target: "rpc::engine", "Serving engine_getBlobsV3");
        Ok(self.get_blobs_v3_metered(versioned_hashes)?)
    }

    async fn get_blobs_v4(
        &self,
        versioned_hashes: Vec<B256>,
        indices_bitarray: B128,
    ) -> RpcResult<Option<Vec<Option<BlobCellsAndProofsV1>>>> {
        trace!(target: "rpc::engine", "Serving engine_getBlobsV4");
        Ok(self.get_blobs_v4_metered(versioned_hashes, indices_bitarray)?)
    }
}

impl<Provider, EngineT, Pool, Validator, ChainSpec> IntoEngineApiRpcModule
    for EngineApi<Provider, EngineT, Pool, Validator, ChainSpec>
where
    EngineT: EngineTypes,
    Self: EngineApiServer<EngineT>,
{
    fn into_rpc_module(self) -> RpcModule<()> {
        EngineApiServer::<EngineT>::into_rpc(self).remove_context()
    }
}

impl<Provider, PayloadT, Pool, Validator, ChainSpec> std::fmt::Debug
    for EngineApi<Provider, PayloadT, Pool, Validator, ChainSpec>
where
    PayloadT: PayloadTypes,
{
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("EngineApi").finish_non_exhaustive()
    }
}

impl<Provider, PayloadT, Pool, Validator, ChainSpec> Clone
    for EngineApi<Provider, PayloadT, Pool, Validator, ChainSpec>
where
    PayloadT: PayloadTypes,
{
    fn clone(&self) -> Self {
        Self { inner: Arc::clone(&self.inner) }
    }
}

/// The container type for the engine API internals.
struct EngineApiInner<Provider, PayloadT: PayloadTypes, Pool, Validator, ChainSpec> {
    /// The provider to interact with the chain.
    provider: Provider,
    /// Consensus configuration
    chain_spec: Arc<ChainSpec>,
    /// The channel to send messages to the beacon consensus engine.
    beacon_consensus: ConsensusEngineHandle<PayloadT>,
    /// The type that can communicate with the payload service to retrieve payloads.
    payload_store: PayloadStore<PayloadT>,
    /// For spawning and executing async tasks
    task_spawner: Runtime,
    /// The latency and response type metrics for engine api calls
    metrics: EngineApiMetrics,
    /// Identification of the execution client used by the consensus client
    client: ClientVersionV1,
    /// The list of all supported Engine capabilities available over the engine endpoint.
    capabilities: EngineCapabilities,
    /// Transaction pool.
    tx_pool: Pool,
    /// Engine validator.
    validator: Validator,
    accept_execution_requests_hash: bool,
    /// Shared blob cell custody bitmap.
    cell_custody: CellCustody,
    /// Returns `true` if the node is currently syncing.
    is_syncing: Arc<dyn Fn() -> bool + Send + Sync>,
}

#[cfg(test)]
mod tests {
    use super::*;
    use alloy_eips::{eip7685::Requests, Encodable2718, NumHash};
    use alloy_primitives::{Address, Bytes, B256};
    use alloy_rpc_types_engine::{
        ClientCode, ClientVersionV1, ExecutionPayloadV2, PayloadAttributes, PayloadStatusEnum,
    };
    use assert_matches::assert_matches;
    use reth_chainspec::{ChainSpec, ChainSpecBuilder, MAINNET};
    use reth_engine_primitives::{BeaconEngineMessage, OnForkChoiceUpdated};
    use reth_ethereum_engine_primitives::EthEngineTypes;
    use reth_ethereum_primitives::{Block, TransactionSigned};
    use reth_network_api::{
        noop::NoopNetwork, EthProtocolInfo, NetworkError, NetworkInfo, NetworkStatus,
    };
    use reth_node_ethereum::EthereumEngineValidator;
    use reth_payload_builder::test_utils::spawn_test_payload_service;
    use reth_primitives_traits::SignedTransaction;
    use reth_provider::{test_utils::MockEthProvider, BalStoreHandle, InMemoryBalStore, RawBal};
    use reth_tasks::Runtime;
    use reth_transaction_pool::{
        blobstore::InMemoryBlobStore,
        noop::NoopTransactionPool,
        test_utils::{OkValidator, TransactionBuilder},
        CoinbaseTipOrdering, EthPooledTransaction, Pool, PoolTransaction, TransactionOrigin,
    };
    use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver};

    type EthTestPool = Pool<
        OkValidator<EthPooledTransaction>,
        CoinbaseTipOrdering<EthPooledTransaction>,
        InMemoryBlobStore,
    >;

    fn eth_test_pool() -> EthTestPool {
        Pool::new(
            OkValidator::default(),
            CoinbaseTipOrdering::default(),
            InMemoryBlobStore::default(),
            Default::default(),
        )
    }

    fn pooled_transaction(transaction: TransactionSigned) -> EthPooledTransaction {
        let transaction = transaction.try_into_recovered().unwrap();
        let encoded_length = transaction.encode_2718_len();
        EthPooledTransaction::new(transaction, encoded_length)
    }

    fn setup_engine_api() -> (
        EngineApiTestHandle,
        EngineApi<
            Arc<MockEthProvider>,
            EthEngineTypes,
            NoopTransactionPool,
            EthereumEngineValidator,
            ChainSpec,
        >,
    ) {
        setup_engine_api_with_pool(NoopTransactionPool::default())
    }

    fn setup_engine_api_with_pool<Pool>(
        tx_pool: Pool,
    ) -> (
        EngineApiTestHandle,
        EngineApi<Arc<MockEthProvider>, EthEngineTypes, Pool, EthereumEngineValidator, ChainSpec>,
    )
    where
        Pool: TransactionPool + 'static,
    {
        setup_engine_api_with_provider(tx_pool, MockEthProvider::default())
    }

    fn setup_engine_api_with_provider<Pool>(
        tx_pool: Pool,
        provider: MockEthProvider,
    ) -> (
        EngineApiTestHandle,
        EngineApi<Arc<MockEthProvider>, EthEngineTypes, Pool, EthereumEngineValidator, ChainSpec>,
    )
    where
        Pool: TransactionPool + 'static,
    {
        let client = ClientVersionV1 {
            code: ClientCode::RH,
            name: "Reth".to_string(),
            version: "v0.2.0-beta.5".to_string(),
            commit: "defa64b2".to_string(),
        };

        let chain_spec: Arc<ChainSpec> = MAINNET.clone();
        let provider = Arc::new(provider);
        let payload_store = spawn_test_payload_service();
        let (to_engine, engine_rx) = unbounded_channel();
        let task_executor = Runtime::test();
        let api = EngineApi::new(
            provider.clone(),
            chain_spec.clone(),
            ConsensusEngineHandle::new(to_engine),
            payload_store.into(),
            tx_pool,
            task_executor,
            client,
            EngineCapabilities::default(),
            EthereumEngineValidator::new(chain_spec.clone()),
            false,
            NoopNetwork::default(),
        );
        let handle = EngineApiTestHandle { chain_spec, provider, from_api: engine_rx };
        (handle, api)
    }

    #[tokio::test]
    async fn engine_client_version_v1() {
        let client = ClientVersionV1 {
            code: ClientCode::RH,
            name: "Reth".to_string(),
            version: "v0.2.0-beta.5".to_string(),
            commit: "defa64b2".to_string(),
        };
        let (_, api) = setup_engine_api();
        let res = api.get_client_version_v1(client.clone());
        assert_eq!(res.unwrap(), vec![client]);
    }

    #[tokio::test]
    async fn get_inclusion_list_v1_returns_empty_list() {
        let (_, api) = setup_engine_api();

        let res = EngineApiServer::get_inclusion_list_v1(&api).await.unwrap();
        assert!(res.is_empty());
    }

    #[tokio::test]
    async fn get_inclusion_list_v1_stops_at_size_limit() {
        let pool = eth_test_pool();
        let first = pooled_transaction(
            TransactionBuilder::default()
                .max_fee_per_gas(3_000_000_000u128)
                .input(vec![0; 4_200])
                .into_legacy(),
        );
        let second = pooled_transaction(
            TransactionBuilder::default()
                .max_fee_per_gas(2_000_000_000u128)
                .input(vec![0; 4_200])
                .into_legacy(),
        );
        let third = pooled_transaction(
            TransactionBuilder::default().max_fee_per_gas(1_000_000_000u128).into_legacy(),
        );
        let expected = first.encoded_2718_consensus();

        pool.add_transaction(TransactionOrigin::External, first).await.unwrap();
        pool.add_transaction(TransactionOrigin::External, second).await.unwrap();
        pool.add_transaction(TransactionOrigin::External, third).await.unwrap();
        let (_, api) = setup_engine_api_with_pool(pool);

        let res = EngineApiServer::get_inclusion_list_v1(&api).await.unwrap();
        assert_eq!(res, vec![expected]);
        assert!(
            res.iter().map(|tx| tx.len()).sum::<usize>() <= MAX_BYTES_PER_INCLUSION_LIST as usize
        );
    }

    /// Signs candidate transactions because signature lengths affect the final encoded size.
    fn inclusion_list_transaction(size: usize, typed: bool, fee: u128) -> EthPooledTransaction {
        for input_size in size.saturating_sub(256)..size {
            let builder = TransactionBuilder::default()
                .signer(B256::with_last_byte(if typed { 2 } else { 1 }))
                .max_fee_per_gas(fee)
                .max_priority_fee_per_gas(fee)
                .input(vec![0; input_size]);
            let transaction = if typed { builder.into_eip1559() } else { builder.into_legacy() };
            if transaction.encode_2718_len() == size {
                return pooled_transaction(transaction)
            }
        }
        panic!("could not construct a transaction of {size} bytes")
    }

    #[tokio::test]
    async fn get_inclusion_list_v1_transaction_byte_boundaries() {
        let limit = MAX_BYTES_PER_INCLUSION_LIST as usize;
        for typed in [false, true] {
            for size in [limit - 3, limit, limit + 1] {
                let pool = eth_test_pool();
                let transaction = inclusion_list_transaction(size, typed, 3_000_000_000);
                let encoded = transaction.encoded_2718_consensus();
                assert_eq!(encoded.len(), size);
                pool.add_transaction(TransactionOrigin::External, transaction).await.unwrap();
                let (_, api) = setup_engine_api_with_pool(pool);

                let res = EngineApiServer::get_inclusion_list_v1(&api).await.unwrap();
                let expected = if size <= limit { vec![encoded] } else { vec![] };
                assert_eq!(res, expected, "typed={typed}, size={size}");
                assert!(res.iter().all(|tx| !tx.is_empty()));
            }
        }
    }

    #[tokio::test]
    async fn get_inclusion_list_v1_cumulative_transaction_byte_boundaries() {
        let limit = MAX_BYTES_PER_INCLUSION_LIST as usize;
        for extra in [0, 1] {
            let pool = eth_test_pool();
            let first = inclusion_list_transaction(limit / 2, false, 3_000_000_000);
            let second = inclusion_list_transaction(limit / 2 + extra, true, 2_000_000_000);
            let mut expected =
                vec![first.encoded_2718_consensus(), second.encoded_2718_consensus()];
            assert_eq!(expected.iter().map(|tx| tx.len()).sum::<usize>(), limit + extra);
            if extra != 0 {
                expected.pop();
            }
            pool.add_transaction(TransactionOrigin::External, first).await.unwrap();
            pool.add_transaction(TransactionOrigin::External, second).await.unwrap();
            let (_, api) = setup_engine_api_with_pool(pool);

            let res = EngineApiServer::get_inclusion_list_v1(&api).await.unwrap();
            assert_eq!(res, expected);
            assert!(res.iter().all(|tx| !tx.is_empty()));
        }
    }

    #[tokio::test]
    async fn get_inclusion_list_v1_excludes_blob_transactions() {
        let pool = eth_test_pool();
        let blob = pooled_transaction(
            TransactionBuilder::default()
                .max_fee_per_gas(2_000_000_000u128)
                .max_priority_fee_per_gas(1_000_000_000u128)
                .into_eip4844(),
        );
        let non_blob = pooled_transaction(
            TransactionBuilder::default()
                .max_fee_per_gas(1_000_000_000u128)
                .max_priority_fee_per_gas(1_000_000_000u128)
                .into_eip1559(),
        );
        let expected = non_blob.encoded_2718_consensus();

        pool.add_transaction(TransactionOrigin::External, blob).await.unwrap();
        pool.add_transaction(TransactionOrigin::External, non_blob).await.unwrap();
        let (_, api) = setup_engine_api_with_pool(pool);

        let res = EngineApiServer::get_inclusion_list_v1(&api).await.unwrap();
        assert_eq!(res, vec![expected]);
    }

    #[tokio::test]
    async fn has_blobs_returns_ordered_availability() {
        let (_, api) = setup_engine_api();

        let res = api.has_blobs_metered(vec![B256::ZERO, B256::with_last_byte(1)]).unwrap();
        assert_eq!(res, vec![false, false]);
    }

    #[tokio::test]
    async fn has_blobs_rejects_large_requests() {
        let (_, api) = setup_engine_api();

        let res = api.has_blobs_metered(vec![B256::ZERO; MAX_BLOB_LIMIT + 1]);
        assert_matches!(
            res,
            Err(EngineApiError::BlobRequestTooLarge { len }) if len == MAX_BLOB_LIMIT + 1
        );
    }

    #[tokio::test]
    async fn payload_bodies_with_timestamps_preserve_missing_entries_and_head_truncation() {
        let mut provider = MockEthProvider::default();
        provider.bal_store = BalStoreHandle::new(InMemoryBalStore::default());
        let (handle, api) =
            setup_engine_api_with_provider(NoopTransactionPool::default(), provider);
        let mut block = Block::default();
        block.header.number = 1;
        block.header.timestamp = 123;
        let hash = block.header.hash_slow();
        handle.provider.add_block(hash, block);
        let raw_bal = Bytes::from_static(&[alloy_rlp::EMPTY_LIST_CODE]);
        handle
            .provider
            .bal_store
            .insert(NumHash::new(1, hash), RawBal::new(raw_bal.clone()))
            .unwrap();
        for include_bal in [false, true] {
            let by_hash = api
                .get_payload_bodies_by_hash_with_timestamps(
                    vec![B256::ZERO, hash, hash],
                    include_bal,
                )
                .await
                .unwrap();
            assert_eq!(by_hash.len(), 3);
            assert!(by_hash[0].is_none());
            assert_eq!(by_hash[1], by_hash[2]);
            let (timestamp, body) = by_hash[1].as_ref().unwrap();
            assert_eq!(*timestamp, 123);
            assert_eq!(body.block_access_list, include_bal.then(|| raw_bal.clone()));
            let by_range =
                api.get_payload_bodies_by_range_with_timestamps(1, 3, include_bal).await.unwrap();
            assert_eq!(by_range, vec![by_hash[1].clone()]);
            assert!(api
                .get_payload_bodies_by_range_with_timestamps(2, 3, include_bal)
                .await
                .unwrap()
                .is_empty());
        }
    }

    #[tokio::test]
    async fn get_payload_bodies_by_hash_v2_returns_block_access_list_from_store() {
        let bal_store = BalStoreHandle::new(InMemoryBalStore::default());
        let mut provider = MockEthProvider::default();
        provider.bal_store = bal_store.clone();
        let provider = Arc::new(provider);

        let client = ClientVersionV1 {
            code: ClientCode::RH,
            name: "Reth".to_string(),
            version: "v0.2.0-beta.5".to_string(),
            commit: "defa64b2".to_string(),
        };
        let chain_spec: Arc<ChainSpec> = MAINNET.clone();
        let payload_store = spawn_test_payload_service::<EthEngineTypes>();
        let (to_engine, _engine_rx) = unbounded_channel();
        let api = EngineApi::new(
            provider.clone(),
            chain_spec.clone(),
            ConsensusEngineHandle::new(to_engine),
            payload_store.into(),
            NoopTransactionPool::default(),
            Runtime::test(),
            client,
            EngineCapabilities::default(),
            EthereumEngineValidator::new(chain_spec),
            false,
            NoopNetwork::default(),
        );

        let mut block = Block::default();
        block.header.number = 1;
        let block_hash = block.header.hash_slow();
        provider.add_block(block_hash, block);

        let mut block_without_bal = Block::default();
        block_without_bal.header.number = 2;
        let block_without_bal_hash = block_without_bal.header.hash_slow();
        provider.add_block(block_without_bal_hash, block_without_bal);

        let raw_bal = Bytes::from_static(&[alloy_rlp::EMPTY_LIST_CODE]);
        bal_store.insert(NumHash::new(1, block_hash), RawBal::new(raw_bal.clone())).unwrap();

        let missing_hash = B256::with_last_byte(3);
        let response = api
            .get_payload_bodies_by_hash_v2(vec![block_hash, block_without_bal_hash, missing_hash])
            .await
            .unwrap();

        assert_eq!(response.len(), 3);
        assert_eq!(response[0].as_ref().unwrap().block_access_list, Some(raw_bal));
        assert_eq!(response[1].as_ref().unwrap().block_access_list, None);
        assert!(response[2].is_none());
    }

    #[tokio::test]
    async fn get_payload_bodies_by_range_v2_returns_block_access_lists_from_store() {
        let bal_store = BalStoreHandle::new(InMemoryBalStore::default());
        let mut provider = MockEthProvider::default();
        provider.bal_store = bal_store.clone();
        let provider = Arc::new(provider);

        let client = ClientVersionV1 {
            code: ClientCode::RH,
            name: "Reth".to_string(),
            version: "v0.2.0-beta.5".to_string(),
            commit: "defa64b2".to_string(),
        };
        let chain_spec: Arc<ChainSpec> = MAINNET.clone();
        let payload_store = spawn_test_payload_service::<EthEngineTypes>();
        let (to_engine, _engine_rx) = unbounded_channel();
        let api = EngineApi::new(
            provider.clone(),
            chain_spec.clone(),
            ConsensusEngineHandle::new(to_engine),
            payload_store.into(),
            NoopTransactionPool::default(),
            Runtime::test(),
            client,
            EngineCapabilities::default(),
            EthereumEngineValidator::new(chain_spec),
            false,
            NoopNetwork::default(),
        );

        let mut block = Block::default();
        block.header.number = 1;
        let block_hash = block.header.hash_slow();
        provider.add_block(block_hash, block);

        let mut block_without_bal = Block::default();
        block_without_bal.header.number = 3;
        let block_without_bal_hash = block_without_bal.header.hash_slow();
        provider.add_block(block_without_bal_hash, block_without_bal);

        let raw_bal = Bytes::from_static(&[alloy_rlp::EMPTY_LIST_CODE]);
        bal_store.insert(NumHash::new(1, block_hash), RawBal::new(raw_bal.clone())).unwrap();

        let response = api.get_payload_bodies_by_range_v2(1, 3).await.unwrap();

        assert_eq!(response.len(), 3);
        assert_eq!(response[0].as_ref().unwrap().block_access_list, Some(raw_bal));
        assert!(response[1].is_none());
        assert_eq!(response[2].as_ref().unwrap().block_access_list, None);
    }

    struct EngineApiTestHandle {
        #[allow(dead_code)]
        chain_spec: Arc<ChainSpec>,
        provider: Arc<MockEthProvider>,
        from_api: UnboundedReceiver<BeaconEngineMessage<EthEngineTypes>>,
    }

    #[tokio::test]
    async fn forwards_responses_to_consensus_engine() {
        let (mut handle, api) = setup_engine_api();

        tokio::spawn(async move {
            let payload_v1 = ExecutionPayloadV1::from_block_slow(&Block::default());
            let execution_data = ExecutionData {
                payload: payload_v1.into(),
                sidecar: ExecutionPayloadSidecar::none(),
            };

            api.new_payload_v1(execution_data).await.unwrap();
        });
        assert_matches!(handle.from_api.recv().await, Some(BeaconEngineMessage::NewPayload { .. }));
    }

    #[tokio::test]
    async fn new_payload_v5_accepts_amsterdam_payloads() {
        let chain_spec = Arc::new(ChainSpecBuilder::mainnet().amsterdam_activated().build());
        let provider = Arc::new(MockEthProvider::default());
        let payload_store = spawn_test_payload_service::<EthEngineTypes>();
        let (to_engine, mut engine_rx) = unbounded_channel();

        let api = EngineApi::new(
            provider,
            chain_spec.clone(),
            ConsensusEngineHandle::new(to_engine),
            payload_store.into(),
            NoopTransactionPool::default(),
            Runtime::test(),
            ClientVersionV1 {
                code: ClientCode::RH,
                name: "Reth".to_string(),
                version: "v0.0.0-test".to_string(),
                commit: "test".to_string(),
            },
            EngineCapabilities::default(),
            EthereumEngineValidator::new(chain_spec),
            false,
            NoopNetwork::default(),
        );

        tokio::spawn(async move {
            let payload_v1 = ExecutionPayloadV1::from_block_slow(&Block::default());
            let payload = ExecutionPayloadV4 {
                payload_inner: ExecutionPayloadV3 {
                    payload_inner: ExecutionPayloadV2 {
                        payload_inner: payload_v1,
                        withdrawals: Vec::new(),
                    },
                    blob_gas_used: 0,
                    excess_blob_gas: 0,
                },
                block_access_list: Bytes::from_static(b"bal"),
                slot_number: 1,
            };
            let execution_data = ExecutionData {
                payload: payload.into(),
                sidecar: ExecutionPayloadSidecar::v4(
                    CancunPayloadFields {
                        versioned_hashes: Vec::new(),
                        parent_beacon_block_root: B256::ZERO,
                    },
                    PraguePayloadFields { requests: RequestsOrHash::Requests(Requests::default()) },
                ),
            };

            api.new_payload_v5(execution_data).await.unwrap();
        });

        assert_matches!(engine_rx.recv().await, Some(BeaconEngineMessage::NewPayload { .. }));
    }

    #[derive(Clone)]
    struct TestNetworkInfo {
        syncing: bool,
    }

    impl NetworkInfo for TestNetworkInfo {
        fn local_addr(&self) -> std::net::SocketAddr {
            (std::net::Ipv4Addr::UNSPECIFIED, 0).into()
        }

        async fn network_status(&self) -> Result<NetworkStatus, NetworkError> {
            #[allow(deprecated)]
            Ok(NetworkStatus {
                client_version: "test".to_string(),
                protocol_version: 5,
                eth_protocol_info: EthProtocolInfo {
                    network: 1,
                    difficulty: None,
                    genesis: Default::default(),
                    config: Default::default(),
                    head: Default::default(),
                },
                capabilities: vec![],
            })
        }

        fn chain_id(&self) -> u64 {
            1
        }

        fn cell_custody(&self) -> &CellCustody {
            static CELL_CUSTODY: std::sync::OnceLock<CellCustody> = std::sync::OnceLock::new();
            CELL_CUSTODY.get_or_init(CellCustody::default)
        }

        fn is_syncing(&self) -> bool {
            self.syncing
        }

        fn is_initially_syncing(&self) -> bool {
            self.syncing
        }
    }

    #[tokio::test]
    async fn get_blobs_v3_returns_null_when_syncing() {
        let chain_spec: Arc<ChainSpec> =
            Arc::new(ChainSpecBuilder::mainnet().osaka_activated().build());
        let provider = Arc::new(MockEthProvider::default());
        let payload_store = spawn_test_payload_service::<EthEngineTypes>();
        let (to_engine, _engine_rx) = unbounded_channel::<BeaconEngineMessage<EthEngineTypes>>();

        let api = EngineApi::new(
            provider,
            chain_spec.clone(),
            ConsensusEngineHandle::new(to_engine),
            payload_store.into(),
            NoopTransactionPool::default(),
            Runtime::test(),
            ClientVersionV1 {
                code: ClientCode::RH,
                name: "Reth".to_string(),
                version: "v0.0.0-test".to_string(),
                commit: "test".to_string(),
            },
            EngineCapabilities::default(),
            EthereumEngineValidator::new(chain_spec),
            false,
            TestNetworkInfo { syncing: true },
        );

        let res = api.get_blobs_v3_metered(vec![B256::ZERO]);
        assert_matches!(res, Ok(None));
    }

    #[tokio::test]
    async fn get_blobs_v4_returns_null_when_syncing() {
        let chain_spec: Arc<ChainSpec> =
            Arc::new(ChainSpecBuilder::mainnet().osaka_activated().build());
        let provider = Arc::new(MockEthProvider::default());
        let payload_store = spawn_test_payload_service::<EthEngineTypes>();
        let (to_engine, _engine_rx) = unbounded_channel::<BeaconEngineMessage<EthEngineTypes>>();

        let api = EngineApi::new(
            provider,
            chain_spec.clone(),
            ConsensusEngineHandle::new(to_engine),
            payload_store.into(),
            NoopTransactionPool::default(),
            Runtime::test(),
            ClientVersionV1 {
                code: ClientCode::RH,
                name: "Reth".to_string(),
                version: "v0.0.0-test".to_string(),
                commit: "test".to_string(),
            },
            EngineCapabilities::default(),
            EthereumEngineValidator::new(chain_spec),
            false,
            TestNetworkInfo { syncing: true },
        );

        let res = api.get_blobs_v4_metered(vec![B256::ZERO], B128::from(1u128.to_le_bytes()));
        assert_matches!(res, Ok(None));
    }

    #[test]
    fn engine_bitvector_uses_little_endian_cell_indices() {
        assert_eq!(
            B128::from(u128::from_le_bytes(B128::from(1u128.to_le_bytes()).into())),
            B128::from(1u128)
        );
        assert_eq!(
            B128::from(u128::from_le_bytes(B128::from((1u128 << 127).to_le_bytes()).into())),
            B128::from(1u128 << 127)
        );
        assert_eq!(
            B128::from(u128::from_le_bytes(B128::from(((1u128 << 64) - 1).to_le_bytes()).into(),)),
            B128::from((1u128 << 64) - 1)
        );
    }

    #[tokio::test]
    async fn fcu_v4_updates_shared_cell_custody_before_forkchoice_result() {
        let chain_spec: Arc<ChainSpec> =
            Arc::new(ChainSpecBuilder::mainnet().amsterdam_activated().build());
        let provider = Arc::new(MockEthProvider::default());
        let payload_store = spawn_test_payload_service::<EthEngineTypes>();
        let (to_engine, mut engine_rx) = unbounded_channel();
        let network = NoopNetwork::default();
        let cell_custody = network.cell_custody().clone();

        let api = EngineApi::new(
            provider,
            chain_spec.clone(),
            ConsensusEngineHandle::new(to_engine),
            payload_store.into(),
            NoopTransactionPool::default(),
            Runtime::test(),
            ClientVersionV1 {
                code: ClientCode::RH,
                name: "Reth".to_string(),
                version: "v0.0.0-test".to_string(),
                commit: "test".to_string(),
            },
            EngineCapabilities::default(),
            EthereumEngineValidator::new(chain_spec),
            false,
            network,
        );

        let state = ForkchoiceState {
            head_block_hash: B256::from([0x33; 32]),
            safe_block_hash: B256::ZERO,
            finalized_block_hash: B256::ZERO,
        };
        let custody_columns = B128::from(0b1010u128.to_le_bytes());
        let expected_custody_columns = B128::from(0b1010u128);

        let api_task = tokio::spawn(async move {
            api.fork_choice_updated_v4(state, None, Some(custody_columns)).await
        });

        let request = tokio::time::timeout(std::time::Duration::from_secs(1), engine_rx.recv())
            .await
            .expect("timed out waiting for forkchoiceUpdated request")
            .expect("expected forkchoiceUpdated request");
        let response_tx = match request {
            BeaconEngineMessage::ForkchoiceUpdated { payload_attrs, tx, .. } => {
                assert!(payload_attrs.is_none());
                tx
            }
            other => panic!("unexpected engine message: {other:?}"),
        };
        assert_eq!(cell_custody.get(), expected_custody_columns);

        response_tx
            .send(Ok(OnForkChoiceUpdated::valid(PayloadStatus::from_status(
                PayloadStatusEnum::Valid,
            ))))
            .expect("send valid response");

        api_task
            .await
            .expect("api task should not panic")
            .expect("forkchoiceUpdatedV4 should succeed");
        assert_eq!(cell_custody.get(), expected_custody_columns);
    }

    #[tokio::test]
    async fn fcu_v4_updates_shared_cell_custody_when_payload_attrs_invalid() {
        let chain_spec: Arc<ChainSpec> =
            Arc::new(ChainSpecBuilder::mainnet().amsterdam_activated().build());
        let provider = Arc::new(MockEthProvider::default());
        let payload_store = spawn_test_payload_service::<EthEngineTypes>();
        let (to_engine, mut engine_rx) = unbounded_channel();
        let network = NoopNetwork::default();
        let cell_custody = network.cell_custody().clone();

        let api = EngineApi::new(
            provider,
            chain_spec.clone(),
            ConsensusEngineHandle::new(to_engine),
            payload_store.into(),
            NoopTransactionPool::default(),
            Runtime::test(),
            ClientVersionV1 {
                code: ClientCode::RH,
                name: "Reth".to_string(),
                version: "v0.0.0-test".to_string(),
                commit: "test".to_string(),
            },
            EngineCapabilities::default(),
            EthereumEngineValidator::new(chain_spec),
            false,
            network,
        );

        let state = ForkchoiceState {
            head_block_hash: B256::from([0x44; 32]),
            safe_block_hash: B256::ZERO,
            finalized_block_hash: B256::ZERO,
        };
        let payload_attributes = PayloadAttributes {
            timestamp: 1,
            prev_randao: B256::ZERO,
            suggested_fee_recipient: Address::ZERO,
            withdrawals: Some(vec![]),
            parent_beacon_block_root: None,
            slot_number: None,
            ..Default::default()
        };
        let custody_columns = B128::from(0b1010u128.to_le_bytes());
        let expected_custody_columns = B128::from(0b1010u128);

        let api_task = tokio::spawn(async move {
            api.fork_choice_updated_v4(state, Some(payload_attributes), Some(custody_columns)).await
        });

        let request = tokio::time::timeout(std::time::Duration::from_secs(1), engine_rx.recv())
            .await
            .expect("timed out waiting for forkchoiceUpdated request")
            .expect("expected forkchoiceUpdated request");
        let response_tx = match request {
            BeaconEngineMessage::ForkchoiceUpdated { payload_attrs, tx, .. } => {
                assert!(
                    payload_attrs.is_none(),
                    "when attrs are invalid, API should first evaluate forkchoice without attrs"
                );
                tx
            }
            other => panic!("unexpected engine message: {other:?}"),
        };
        assert_eq!(cell_custody.get(), expected_custody_columns);

        response_tx
            .send(Ok(OnForkChoiceUpdated::valid(PayloadStatus::from_status(
                PayloadStatusEnum::Valid,
            ))))
            .expect("send valid response");

        let response = api_task.await.expect("api task should not panic");
        assert_matches!(
            response,
            Err(EngineApiError::EngineObjectValidationError(
                reth_payload_primitives::EngineObjectValidationError::PayloadAttributes(_)
            ))
        );
    }

    #[tokio::test]
    async fn fcu_v3_syncing_precedes_invalid_payload_attributes_validation() {
        let (mut handle, api) = setup_engine_api();

        let state = ForkchoiceState {
            head_block_hash: B256::from([0x11; 32]),
            safe_block_hash: B256::ZERO,
            finalized_block_hash: B256::ZERO,
        };
        let payload_attributes = PayloadAttributes {
            timestamp: 1,
            prev_randao: B256::ZERO,
            suggested_fee_recipient: Address::ZERO,
            withdrawals: Some(vec![]),
            // Invalid for V3/Cancun, but should be ignored if forkchoice is SYNCING.
            parent_beacon_block_root: None,
            slot_number: None,
            ..Default::default()
        };

        let api_task = tokio::spawn(async move {
            api.fork_choice_updated_v3(state, Some(payload_attributes)).await
        });

        let request =
            tokio::time::timeout(std::time::Duration::from_secs(1), handle.from_api.recv())
                .await
                .expect("timed out waiting for forkchoiceUpdated request")
                .expect("expected forkchoiceUpdated request");
        let response_tx = match request {
            BeaconEngineMessage::ForkchoiceUpdated { payload_attrs, tx, .. } => {
                assert!(
                    payload_attrs.is_none(),
                    "FCU for syncing state should be evaluated before payload attributes"
                );
                tx
            }
            other => panic!("unexpected engine message: {other:?}"),
        };

        response_tx.send(Ok(OnForkChoiceUpdated::syncing())).expect("send syncing response");

        let response = api_task
            .await
            .expect("api task should not panic")
            .expect("forkchoiceUpdatedV3 should return a syncing response");
        assert!(response.payload_status.is_syncing());
        assert!(response.payload_id.is_none());
    }

    #[tokio::test]
    async fn fcu_v3_valid_forkchoice_missing_beacon_root_returns_invalid_attributes() {
        let (mut handle, api) = setup_engine_api();

        let state = ForkchoiceState {
            head_block_hash: B256::from([0x22; 32]),
            safe_block_hash: B256::ZERO,
            finalized_block_hash: B256::ZERO,
        };
        let payload_attributes = PayloadAttributes {
            timestamp: 1,
            prev_randao: B256::ZERO,
            suggested_fee_recipient: Address::ZERO,
            withdrawals: Some(vec![]),
            parent_beacon_block_root: None,
            slot_number: None,
            ..Default::default()
        };

        let api_task = tokio::spawn(async move {
            api.fork_choice_updated_v3(state, Some(payload_attributes)).await
        });

        let request =
            tokio::time::timeout(std::time::Duration::from_secs(1), handle.from_api.recv())
                .await
                .expect("timed out waiting for forkchoiceUpdated request")
                .expect("expected forkchoiceUpdated request");

        let response_tx = match request {
            BeaconEngineMessage::ForkchoiceUpdated { payload_attrs, tx, .. } => {
                assert!(
                    payload_attrs.is_none(),
                    "when attrs are invalid, API should first evaluate forkchoice without attrs"
                );
                tx
            }
            other => panic!("unexpected engine message: {other:?}"),
        };

        response_tx
            .send(Ok(OnForkChoiceUpdated::valid(PayloadStatus::from_status(
                PayloadStatusEnum::Valid,
            ))))
            .expect("send valid response");

        let response = api_task.await.expect("api task should not panic");
        assert_matches!(
            response,
            Err(EngineApiError::EngineObjectValidationError(
                reth_payload_primitives::EngineObjectValidationError::PayloadAttributes(_)
            ))
        );

        match tokio::time::timeout(std::time::Duration::from_millis(100), handle.from_api.recv())
            .await
        {
            Err(_) | Ok(None) => {}
            Ok(Some(BeaconEngineMessage::ForkchoiceUpdated { .. })) => {
                panic!("no second forkchoiceUpdated call should be sent when attrs are invalid")
            }
            Ok(Some(other)) => panic!("unexpected engine message: {other:?}"),
        }
    }

    // tests covering `engine_getPayloadBodiesByRange` and `engine_getPayloadBodiesByHash`
    mod get_payload_bodies {
        use super::*;
        use alloy_rpc_types_engine::ExecutionPayloadBodyV1;
        use reth_testing_utils::generators::{self, random_block_range, BlockRangeParams};

        #[tokio::test]
        async fn invalid_params() {
            let (_, api) = setup_engine_api();

            let by_range_tests = [
                // (start, count)
                (0, 0),
                (0, 1),
                (1, 0),
            ];

            // test [EngineApiMessage::GetPayloadBodiesByRange]
            for (start, count) in by_range_tests {
                let res = api.get_payload_bodies_by_range_v1(start, count).await;
                assert_matches!(res, Err(EngineApiError::InvalidBodiesRange { .. }));
            }
        }

        #[tokio::test]
        async fn request_too_large() {
            let (_, api) = setup_engine_api();

            let request_count = MAX_PAYLOAD_BODIES_LIMIT + 1;
            let res = api.get_payload_bodies_by_range_v1(0, request_count).await;
            assert_matches!(res, Err(EngineApiError::PayloadRequestTooLarge { .. }));
        }

        #[tokio::test]
        async fn returns_payload_bodies() {
            let mut rng = generators::rng();
            let (handle, api) = setup_engine_api();

            let (start, count) = (1, 10);
            let blocks = random_block_range(
                &mut rng,
                start..=start + count - 1,
                BlockRangeParams { tx_count: 0..2, ..Default::default() },
            );
            handle
                .provider
                .extend_blocks(blocks.iter().cloned().map(|b| (b.hash(), b.into_block())));

            let expected = blocks
                .iter()
                .cloned()
                .map(|b| Some(ExecutionPayloadBodyV1::from_block(b.into_block())))
                .collect::<Vec<_>>();

            let res = api.get_payload_bodies_by_range_v1(start, count).await.unwrap();
            assert_eq!(res, expected);
        }

        #[tokio::test]
        async fn returns_payload_bodies_with_gaps() {
            let mut rng = generators::rng();
            let (handle, api) = setup_engine_api();

            let (start, count) = (1, 100);
            let blocks = random_block_range(
                &mut rng,
                start..=start + count - 1,
                BlockRangeParams { tx_count: 0..2, ..Default::default() },
            );

            // Insert only blocks in ranges 1-25 and 50-75
            let first_missing_range = 26..=50;
            let second_missing_range = 76..=100;
            handle.provider.extend_blocks(
                blocks
                    .iter()
                    .filter(|b| {
                        !first_missing_range.contains(&b.number) &&
                            !second_missing_range.contains(&b.number)
                    })
                    .map(|b| (b.hash(), b.clone().into_block())),
            );

            let expected = blocks
                .iter()
                // filter anything after the second missing range to ensure we don't expect trailing
                // `None`s
                .filter(|b| !second_missing_range.contains(&b.number))
                .cloned()
                .map(|b| {
                    if first_missing_range.contains(&b.number) {
                        None
                    } else {
                        Some(ExecutionPayloadBodyV1::from_block(b.into_block()))
                    }
                })
                .collect::<Vec<_>>();

            let res = api.get_payload_bodies_by_range_v1(start, count).await.unwrap();
            assert_eq!(res, expected);

            let expected = blocks
                .iter()
                .cloned()
                // ensure we still return trailing `None`s here because by-hash will not be aware
                // of the missing block's number, and cannot compare it to the current best block
                .map(|b| {
                    if first_missing_range.contains(&b.number) ||
                        second_missing_range.contains(&b.number)
                    {
                        None
                    } else {
                        Some(ExecutionPayloadBodyV1::from_block(b.into_block()))
                    }
                })
                .collect::<Vec<_>>();

            let hashes = blocks.iter().map(|b| b.hash()).collect();
            let res = api.get_payload_bodies_by_hash_v1(hashes).await.unwrap();
            assert_eq!(res, expected);
        }
    }
}