Use efficient payload reconstruction for HTTP API (sigp#4102)

## Proposed Changes

Builds on sigp#4028 to use the new payload bodies methods in the HTTP API as well.

## Caveats

The payloads by range method only works for the finalized chain, so it can't be used in the execution engine integration tests because we try to reconstruct unfinalized payloads there.

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -1043,7 +1043,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             .execution_layer
             .as_ref()
             .ok_or(Error::ExecutionLayerMissing)?
-            .get_payload_by_block_hash(exec_block_hash, fork)
+            .get_payload_for_header(&execution_payload_header, fork)
             .await
             .map_err(|e| {
                 Error::ExecutionLayerErrorPayloadReconstruction(exec_block_hash, Box::new(e))

beacon_node/execution_layer/src/lib.rs

@@ -103,6 +103,8 @@ pub enum Error {
         transactions_root: Hash256,
     },
     InvalidJWTSecret(String),
+    InvalidForkForPayload,
+    InvalidPayloadBody(String),
     BeaconStateError(BeaconStateError),
 }
 
@@ -1602,22 +1604,67 @@ impl<T: EthSpec> ExecutionLayer<T> {
             .map_err(Error::EngineError)
     }
 
-    pub async fn get_payload_by_block_hash(
+    /// Fetch a full payload from the execution node.
+    ///
+    /// This will fail if the payload is not from the finalized portion of the chain.
+    pub async fn get_payload_for_header(
+        &self,
+        header: &ExecutionPayloadHeader<T>,
+        fork: ForkName,
+    ) -> Result<Option<ExecutionPayload<T>>, Error> {
+        let hash = header.block_hash();
+        let block_number = header.block_number();
+
+        // Handle default payload body.
+        if header.block_hash() == ExecutionBlockHash::zero() {
+            let payload = match fork {
+                ForkName::Merge => ExecutionPayloadMerge::default().into(),
+                ForkName::Capella => ExecutionPayloadCapella::default().into(),
+                ForkName::Base | ForkName::Altair => {
+                    return Err(Error::InvalidForkForPayload);
+                }
+            };
+            return Ok(Some(payload));
+        }
+
+        // Use efficient payload bodies by range method if supported.
+        let capabilities = self.get_engine_capabilities(None).await?;
+        if capabilities.get_payload_bodies_by_range_v1 {
+            let mut payload_bodies = self.get_payload_bodies_by_range(block_number, 1).await?;
+
+            if payload_bodies.len() != 1 {
+                return Ok(None);
+            }
+
+            let opt_payload_body = payload_bodies.pop().flatten();
+            opt_payload_body
+                .map(|body| {
+                    body.to_payload(header.clone())
+                        .map_err(Error::InvalidPayloadBody)
+                })
+                .transpose()
+        } else {
+            // Fall back to eth_blockByHash.
+            self.get_payload_by_hash_legacy(hash, fork).await
+        }
+    }
+
+    pub async fn get_payload_by_hash_legacy(
         &self,
         hash: ExecutionBlockHash,
         fork: ForkName,
     ) -> Result<Option<ExecutionPayload<T>>, Error> {
         self.engine()
             .request(|engine| async move {
-                self.get_payload_by_block_hash_from_engine(engine, hash, fork)
+                self.get_payload_by_hash_from_engine(engine, hash, fork)
                     .await
             })
             .await
             .map_err(Box::new)
             .map_err(Error::EngineError)
     }
 
-    async fn get_payload_by_block_hash_from_engine(
+    async fn get_payload_by_hash_from_engine(
         &self,
         engine: &Engine,
         hash: ExecutionBlockHash,
@@ -1630,7 +1677,7 @@ impl<T: EthSpec> ExecutionLayer<T> {
                 ForkName::Merge => Ok(Some(ExecutionPayloadMerge::default().into())),
                 ForkName::Capella => Ok(Some(ExecutionPayloadCapella::default().into())),
                 ForkName::Base | ForkName::Altair => Err(ApiError::UnsupportedForkVariant(
-                    format!("called get_payload_by_block_hash_from_engine with {}", fork),
+                    format!("called get_payload_by_hash_from_engine with {}", fork),
                 )),
             };
         }

testing/execution_engine_integration/src/test_rig.rs

@@ -626,9 +626,10 @@ async fn check_payload_reconstruction<E: GenericExecutionEngine>(
     ee: &ExecutionPair<E, MainnetEthSpec>,
     payload: &ExecutionPayload<MainnetEthSpec>,
 ) {
+    // check via legacy eth_getBlockByHash
     let reconstructed = ee
         .execution_layer
-        .get_payload_by_block_hash(payload.block_hash(), payload.fork_name())
+        .get_payload_by_hash_legacy(payload.block_hash(), payload.fork_name())
         .await
         .unwrap()
         .unwrap();