feat: add support for payload bodies (paradigmxyz#9378)

crates/rpc/rpc-api/src/engine.rs

@@ -8,14 +8,13 @@ use reth_engine_primitives::EngineTypes;
 use reth_primitives::{Address, BlockHash, BlockId, BlockNumberOrTag, Bytes, B256, U256, U64};
 use reth_rpc_types::{
     engine::{
-        ClientVersionV1, ExecutionPayloadBodiesV1, ExecutionPayloadInputV2, ExecutionPayloadV1,
-        ExecutionPayloadV3, ExecutionPayloadV4, ForkchoiceState, ForkchoiceUpdated, PayloadId,
-        PayloadStatus, TransitionConfiguration,
+        ClientVersionV1, ExecutionPayloadBodiesV1, ExecutionPayloadBodiesV2,
+        ExecutionPayloadInputV2, ExecutionPayloadV1, ExecutionPayloadV3, ExecutionPayloadV4,
+        ForkchoiceState, ForkchoiceUpdated, PayloadId, PayloadStatus, TransitionConfiguration,
     },
     state::StateOverride,
     BlockOverrides, Filter, Log, RichBlock, SyncStatus, TransactionRequest,
 };
-
 // NOTE: We can't use associated types in the `EngineApi` trait because of jsonrpsee, so we use a
 // generic here. It would be nice if the rpc macro would understand which types need to have serde.
 // By default, if the trait has a generic, the rpc macro will add e.g. `Engine: DeserializeOwned` to
@@ -144,6 +143,13 @@ pub trait EngineApi<Engine: EngineTypes> {
         block_hashes: Vec<BlockHash>,
     ) -> RpcResult<ExecutionPayloadBodiesV1>;
 
+    /// See also <https://github.com/ethereum/execution-apis/blob/main/src/engine/prague.md#engine_getpayloadbodiesbyhashv2>
+    #[method(name = "getPayloadBodiesByHashV2")]
+    async fn get_payload_bodies_by_hash_v2(
+        &self,
+        block_hashes: Vec<BlockHash>,
+    ) -> RpcResult<ExecutionPayloadBodiesV2>;
+
     /// 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`
@@ -163,6 +169,16 @@ pub trait EngineApi<Engine: EngineTypes> {
         count: U64,
     ) -> RpcResult<ExecutionPayloadBodiesV1>;
 
+    /// See also <https://github.com/ethereum/execution-apis/blob/main/src/engine/prague.md#engine_getpayloadbodiesbyrangev2>
+    ///
+    /// Similar to `getPayloadBodiesByRangeV1`, but returns [`ExecutionPayloadBodiesV2`]
+    #[method(name = "getPayloadBodiesByRangeV2")]
+    async fn get_payload_bodies_by_range_v2(
+        &self,
+        start: U64,
+        count: U64,
+    ) -> RpcResult<ExecutionPayloadBodiesV2>;
+
     /// See also <https://github.com/ethereum/execution-apis/blob/6709c2a795b707202e93c4f2867fa0bf2640a84f/src/engine/paris.md#engine_exchangetransitionconfigurationv1>
     ///
     /// Note: This method will be deprecated after the cancun hardfork:

crates/rpc/rpc-engine-api/src/engine_api.rs

@@ -10,16 +10,18 @@ use reth_payload_primitives::{
     validate_payload_timestamp, EngineApiMessageVersion, PayloadAttributes,
     PayloadBuilderAttributes, PayloadOrAttributes,
 };
-use reth_primitives::{BlockHash, BlockHashOrNumber, BlockNumber, EthereumHardfork, B256, U64};
+use reth_primitives::{
+    Block, BlockHash, BlockHashOrNumber, BlockNumber, EthereumHardfork, B256, U64,
+};
 use reth_rpc_api::EngineApiServer;
 use reth_rpc_types::engine::{
     CancunPayloadFields, ClientVersionV1, ExecutionPayload, ExecutionPayloadBodiesV1,
-    ExecutionPayloadInputV2, ExecutionPayloadV1, ExecutionPayloadV3, ExecutionPayloadV4,
-    ForkchoiceState, ForkchoiceUpdated, PayloadId, PayloadStatus, TransitionConfiguration,
-    CAPABILITIES,
+    ExecutionPayloadBodiesV2, ExecutionPayloadInputV2, ExecutionPayloadV1, ExecutionPayloadV3,
+    ExecutionPayloadV4, ForkchoiceState, ForkchoiceUpdated, PayloadId, PayloadStatus,
+    TransitionConfiguration, CAPABILITIES,
 };
 use reth_rpc_types_compat::engine::payload::{
-    convert_payload_input_v2_to_payload, convert_to_payload_body_v1,
+    convert_payload_input_v2_to_payload, convert_to_payload_body_v1, convert_to_payload_body_v2,
 };
 use reth_storage_api::{BlockReader, HeaderProvider, StateProviderFactory};
 use reth_tasks::TaskSpawner;
@@ -359,21 +361,18 @@ where
             })
     }
 
-    /// 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(
+    /// Fetches all the blocks for the provided range starting at `start`, containing `count`
+    /// blocks and returns the mapped payload bodies.
+    async fn get_payload_bodies_by_range_with<F, R>(
         &self,
         start: BlockNumber,
         count: u64,
-    ) -> EngineApiResult<ExecutionPayloadBodiesV1> {
+        f: F,
+    ) -> EngineApiResult<Vec<Option<R>>>
+    where
+        F: Fn(Block) -> R + Send + 'static,
+        R: Send + 'static,
+    {
         let (tx, rx) = oneshot::channel();
         let inner = self.inner.clone();
 
@@ -405,7 +404,7 @@ where
                 let block_result = inner.provider.block(BlockHashOrNumber::Number(num));
                 match block_result {
                     Ok(block) => {
-                        result.push(block.map(convert_to_payload_body_v1));
+                        result.push(block.map(&f));
                     }
                     Err(err) => {
                         tx.send(Err(EngineApiError::Internal(Box::new(err)))).ok();
@@ -419,11 +418,45 @@ where
         rx.await.map_err(|err| EngineApiError::Internal(Box::new(err)))?
     }
 
+    /// 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, convert_to_payload_body_v1).await
+    }
+
+    /// Returns the execution payload bodies by the range starting at `start`, containing `count`
+    /// blocks.
+    ///
+    /// Same as [`Self::get_payload_bodies_by_range_v1`] but as [`ExecutionPayloadBodiesV2`].
+    pub async fn get_payload_bodies_by_range_v2(
+        &self,
+        start: BlockNumber,
+        count: u64,
+    ) -> EngineApiResult<ExecutionPayloadBodiesV2> {
+        self.get_payload_bodies_by_range_with(start, count, convert_to_payload_body_v2).await
+    }
+
     /// Called to retrieve execution payload bodies by hashes.
-    pub fn get_payload_bodies_by_hash(
+    fn get_payload_bodies_by_hash_with<F, R>(
         &self,
         hashes: Vec<BlockHash>,
-    ) -> EngineApiResult<ExecutionPayloadBodiesV1> {
+        f: F,
+    ) -> EngineApiResult<Vec<Option<R>>>
+    where
+        F: Fn(Block) -> R,
+    {
         let len = hashes.len() as u64;
         if len > MAX_PAYLOAD_BODIES_LIMIT {
             return Err(EngineApiError::PayloadRequestTooLarge { len })
@@ -436,12 +469,30 @@ where
                 .provider
                 .block(BlockHashOrNumber::Hash(hash))
                 .map_err(|err| EngineApiError::Internal(Box::new(err)))?;
-            result.push(block.map(convert_to_payload_body_v1));
+            result.push(block.map(&f));
         }
 
         Ok(result)
     }
 
+    /// Called to retrieve execution payload bodies by hashes.
+    pub fn get_payload_bodies_by_hash_v1(
+        &self,
+        hashes: Vec<BlockHash>,
+    ) -> EngineApiResult<ExecutionPayloadBodiesV1> {
+        self.get_payload_bodies_by_hash_with(hashes, convert_to_payload_body_v1)
+    }
+
+    /// Called to retrieve execution payload bodies by hashes.
+    ///
+    /// Same as [`Self::get_payload_bodies_by_hash_v1`] but as [`ExecutionPayloadBodiesV2`].
+    pub fn get_payload_bodies_by_hash_v2(
+        &self,
+        hashes: Vec<BlockHash>,
+    ) -> EngineApiResult<ExecutionPayloadBodiesV2> {
+        self.get_payload_bodies_by_hash_with(hashes, convert_to_payload_body_v2)
+    }
+
     /// Called to verify network configuration parameters and ensure that Consensus and Execution
     /// layers are using the latest configuration.
     pub fn exchange_transition_configuration(
@@ -760,11 +811,22 @@ where
     ) -> RpcResult<ExecutionPayloadBodiesV1> {
         trace!(target: "rpc::engine", "Serving engine_getPayloadBodiesByHashV1");
         let start = Instant::now();
-        let res = Self::get_payload_bodies_by_hash(self, block_hashes);
+        let res = Self::get_payload_bodies_by_hash_v1(self, block_hashes);
         self.inner.metrics.latency.get_payload_bodies_by_hash_v1.record(start.elapsed());
         Ok(res?)
     }
 
+    async fn get_payload_bodies_by_hash_v2(
+        &self,
+        block_hashes: Vec<BlockHash>,
+    ) -> RpcResult<ExecutionPayloadBodiesV2> {
+        trace!(target: "rpc::engine", "Serving engine_getPayloadBodiesByHashV2");
+        let start = Instant::now();
+        let res = Self::get_payload_bodies_by_hash_v2(self, block_hashes);
+        self.inner.metrics.latency.get_payload_bodies_by_hash_v2.record(start.elapsed());
+        Ok(res?)
+    }
+
     /// Handler for `engine_getPayloadBodiesByRangeV1`
     ///
     /// See also <https://github.com/ethereum/execution-apis/blob/6452a6b194d7db269bf1dbd087a267251d3cc7f8/src/engine/shanghai.md#engine_getpayloadbodiesbyrangev1>
@@ -788,11 +850,23 @@ where
     ) -> RpcResult<ExecutionPayloadBodiesV1> {
         trace!(target: "rpc::engine", "Serving engine_getPayloadBodiesByRangeV1");
         let start_time = Instant::now();
-        let res = Self::get_payload_bodies_by_range(self, start.to(), count.to()).await;
+        let res = Self::get_payload_bodies_by_range_v1(self, start.to(), count.to()).await;
         self.inner.metrics.latency.get_payload_bodies_by_range_v1.record(start_time.elapsed());
         Ok(res?)
     }
 
+    async fn get_payload_bodies_by_range_v2(
+        &self,
+        start: U64,
+        count: U64,
+    ) -> RpcResult<ExecutionPayloadBodiesV2> {
+        trace!(target: "rpc::engine", "Serving engine_getPayloadBodiesByRangeV2");
+        let start_time = Instant::now();
+        let res = Self::get_payload_bodies_by_range_v2(self, start.to(), count.to()).await;
+        self.inner.metrics.latency.get_payload_bodies_by_range_v2.record(start_time.elapsed());
+        Ok(res?)
+    }
+
     /// Handler for `engine_exchangeTransitionConfigurationV1`
     /// See also <https://github.com/ethereum/execution-apis/blob/3d627c95a4d3510a8187dd02e0250ecb4331d27e/src/engine/paris.md#engine_exchangeTransitionConfigurationV1>
     async fn exchange_transition_configuration(
@@ -929,7 +1003,7 @@ mod tests {
 
             // test [EngineApiMessage::GetPayloadBodiesByRange]
             for (start, count) in by_range_tests {
-                let res = api.get_payload_bodies_by_range(start, count).await;
+                let res = api.get_payload_bodies_by_range_v1(start, count).await;
                 assert_matches!(res, Err(EngineApiError::InvalidBodiesRange { .. }));
             }
         }
@@ -939,7 +1013,7 @@ mod tests {
             let (_, api) = setup_engine_api();
 
             let request_count = MAX_PAYLOAD_BODIES_LIMIT + 1;
-            let res = api.get_payload_bodies_by_range(0, request_count).await;
+            let res = api.get_payload_bodies_by_range_v1(0, request_count).await;
             assert_matches!(res, Err(EngineApiError::PayloadRequestTooLarge { .. }));
         }
 
@@ -959,7 +1033,7 @@ mod tests {
                 .map(|b| Some(convert_to_payload_body_v1(b.unseal())))
                 .collect::<Vec<_>>();
 
-            let res = api.get_payload_bodies_by_range(start, count).await.unwrap();
+            let res = api.get_payload_bodies_by_range_v1(start, count).await.unwrap();
             assert_eq!(res, expected);
         }
 
@@ -1000,7 +1074,7 @@ mod tests {
                 })
                 .collect::<Vec<_>>();
 
-            let res = api.get_payload_bodies_by_range(start, count).await.unwrap();
+            let res = api.get_payload_bodies_by_range_v1(start, count).await.unwrap();
             assert_eq!(res, expected);
 
             let expected = blocks
@@ -1020,7 +1094,7 @@ mod tests {
                 .collect::<Vec<_>>();
 
             let hashes = blocks.iter().map(|b| b.hash()).collect();
-            let res = api.get_payload_bodies_by_hash(hashes).unwrap();
+            let res = api.get_payload_bodies_by_hash_v1(hashes).unwrap();
             assert_eq!(res, expected);
         }
     }

crates/rpc/rpc-engine-api/src/metrics.rs

@@ -44,8 +44,12 @@ pub(crate) struct EngineApiLatencyMetrics {
     pub(crate) get_payload_v4: Histogram,
     /// Latency for `engine_getPayloadBodiesByRangeV1`
     pub(crate) get_payload_bodies_by_range_v1: Histogram,
+    /// Latency for `engine_getPayloadBodiesByRangeV2`
+    pub(crate) get_payload_bodies_by_range_v2: Histogram,
     /// Latency for `engine_getPayloadBodiesByHashV1`
     pub(crate) get_payload_bodies_by_hash_v1: Histogram,
+    /// Latency for `engine_getPayloadBodiesByHashV2`
+    pub(crate) get_payload_bodies_by_hash_v2: Histogram,
     /// Latency for `engine_exchangeTransitionConfigurationV1`
     pub(crate) exchange_transition_configuration: Histogram,
 }

crates/rpc/rpc-types-compat/src/engine/payload.rs

@@ -8,8 +8,8 @@ use reth_primitives::{
 };
 use reth_rpc_types::engine::{
     payload::{ExecutionPayloadBodyV1, ExecutionPayloadFieldV2, ExecutionPayloadInputV2},
-    ExecutionPayload, ExecutionPayloadV1, ExecutionPayloadV2, ExecutionPayloadV3,
-    ExecutionPayloadV4, PayloadError,
+    ExecutionPayload, ExecutionPayloadBodyV2, ExecutionPayloadV1, ExecutionPayloadV2,
+    ExecutionPayloadV3, ExecutionPayloadV4, PayloadError,
 };
 
 /// Converts [`ExecutionPayloadV1`] to [Block]
@@ -378,6 +378,52 @@ pub fn convert_to_payload_body_v1(value: Block) -> ExecutionPayloadBodyV1 {
     }
 }
 
+/// Converts [Block] to [`ExecutionPayloadBodyV2`]
+pub fn convert_to_payload_body_v2(value: Block) -> ExecutionPayloadBodyV2 {
+    let transactions = value.body.into_iter().map(|tx| {
+        let mut out = Vec::new();
+        tx.encode_enveloped(&mut out);
+        out.into()
+    });
+
+    let mut payload = ExecutionPayloadBodyV2 {
+        transactions: transactions.collect(),
+        withdrawals: value.withdrawals.map(Withdrawals::into_inner),
+        deposit_requests: None,
+        withdrawal_requests: None,
+        consolidation_requests: None,
+    };
+
+    if let Some(requests) = value.requests {
+        let (deposit_requests, withdrawal_requests, consolidation_requests) =
+            requests.into_iter().fold(
+                (Vec::new(), Vec::new(), Vec::new()),
+                |(mut deposits, mut withdrawals, mut consolidation_requests), request| {
+                    match request {
+                        Request::DepositRequest(r) => {
+                            deposits.push(r);
+                        }
+                        Request::WithdrawalRequest(r) => {
+                            withdrawals.push(r);
+                        }
+                        Request::ConsolidationRequest(r) => {
+                            consolidation_requests.push(r);
+                        }
+                        _ => {}
+                    };
+
+                    (deposits, withdrawals, consolidation_requests)
+                },
+            );
+
+        payload.deposit_requests = Some(deposit_requests);
+        payload.withdrawal_requests = Some(withdrawal_requests);
+        payload.consolidation_requests = Some(consolidation_requests);
+    }
+
+    payload
+}
+
 /// Transforms a [`SealedBlock`] into a [`ExecutionPayloadV1`]
 pub fn execution_payload_from_sealed_block(value: SealedBlock) -> ExecutionPayloadV1 {
     let transactions = value.raw_transactions();