Add REST-SSZ Engine API with Heze support

Cargo.toml

@@ -217,7 +217,7 @@ rand = "0.9.0"
 rand_xorshift = "0.4.0"
 rayon = "1.7"
 regex = "1"
-reqwest = { version = "0.12", default-features = false, features = ["blocking", "json", "stream", "rustls-tls", "rustls-tls-native-roots"] }
+reqwest = { version = "0.12", default-features = false, features = ["blocking", "json", "stream", "rustls-tls", "rustls-tls-native-roots", "http2"] }
 ring = "0.17"
 rpds = "0.11"
 rusqlite = { version = "0.38", features = ["bundled"] }

beacon_node/beacon_chain/src/beacon_block_streamer.rs

@@ -1,5 +1,6 @@
 use crate::{BeaconChain, BeaconChainError, BeaconChainTypes, BlockProcessStatus, metrics};
 use execution_layer::{ExecutionLayer, ExecutionPayloadBodyV1};
+use std::collections::HashMap;
 use std::sync::Arc;
 use store::{DatabaseBlock, ExecutionPayloadDeneb};
 use tokio::sync::mpsc::{self, UnboundedSender};
@@ -27,10 +28,6 @@ pub enum Error {
     InvalidPayloadBodiesResponse { expected: usize, received: usize },
 }
 
-/// The engine API only requires execution layers to support payload body requests for at least
-/// 32 blocks, so larger requests are split into chunks of this size.
-const MAX_PAYLOAD_BODIES_PER_REQUEST: usize = 32;
-
 /// A block loaded from the caches or database, which is either complete or requires its
 /// execution payload to be fetched from the execution layer.
 enum LoadedBlock<E: EthSpec> {
@@ -42,16 +39,19 @@ type BlockResult<E> = Result<Option<Arc<SignedBeaconBlock<E>>>, BeaconChainError
 
 // stores the components of a block for future re-construction in a small form
 struct BlockParts<E: EthSpec> {
+    fork: ForkName,
     blinded_block: Box<SignedBlindedBeaconBlock<E>>,
     header: Box<ExecutionPayloadHeader<E>>,
 }
 
 impl<E: EthSpec> BlockParts<E> {
     pub fn new(
+        fork: ForkName,
         blinded: Box<SignedBlindedBeaconBlock<E>>,
         header: ExecutionPayloadHeader<E>,
     ) -> Self {
         Self {
+            fork,
             blinded_block: blinded,
             header: Box::new(header),
         }
@@ -64,6 +64,10 @@ impl<E: EthSpec> BlockParts<E> {
     pub fn block_hash(&self) -> ExecutionBlockHash {
         self.header.block_hash()
     }
+
+    pub fn fork(&self) -> ForkName {
+        self.fork
+    }
 }
 
 fn reconstruct_default_header_block<E: EthSpec>(
@@ -211,37 +215,68 @@ impl<T: BeaconChainTypes> BeaconBlockStreamer<T> {
                                 &self.beacon_chain.spec,
                             ))
                         } else {
-                            LoadedBlock::NeedsPayload(BlockParts::new(Box::new(block), header))
+                            // Carry the fork for single-fork REST body batching.
+                            match block.fork_name(&self.beacon_chain.spec) {
+                                Ok(fork) => LoadedBlock::NeedsPayload(BlockParts::new(
+                                    fork,
+                                    Box::new(block),
+                                    header,
+                                )),
+                                Err(e) => LoadedBlock::Complete(Err(
+                                    BeaconChainError::InconsistentFork(e),
+                                )),
+                            }
                         }
                     }
                 }
             }
         }
     }
 
-    /// Fetch payload bodies for `block_hashes` from the execution layer.
+    /// Fetch payload bodies for `blocks` (hash + fork) from the execution layer.
     ///
-    /// The returned vec has the same length and order as `block_hashes`. A body the execution
-    /// layer doesn't know about is `None`.
+    /// The returned vec has the same length and order as `blocks`. A body the execution layer
+    /// doesn't know about is `None`. Requests are grouped by fork so each REST `/bodies/hash`
+    /// request is single-fork (decodable at one `Eth-Execution-Version`); JSON-RPC ignores the
+    /// fork, so the grouping is a no-op there.
     async fn fetch_payload_bodies(
         &self,
-        block_hashes: Vec<ExecutionBlockHash>,
+        blocks: Vec<(ExecutionBlockHash, ForkName)>,
     ) -> Result<Vec<Option<ExecutionPayloadBodyV1<T::EthSpec>>>, Error> {
-        let mut bodies = Vec::with_capacity(block_hashes.len());
-        for chunk in block_hashes.chunks(MAX_PAYLOAD_BODIES_PER_REQUEST) {
-            let chunk_bodies = self
-                .execution_layer
-                .get_payload_bodies_by_hash(chunk.to_vec())
-                .await
-                .map_err(|e| Error::BlocksByHashFailure(Box::new(e)))?;
+        // Result slots by original index, so the returned order matches `blocks`.
+        let mut bodies: Vec<Option<ExecutionPayloadBodyV1<T::EthSpec>>> =
+            (0..blocks.len()).map(|_| None).collect();
+
+        let mut by_fork: HashMap<ForkName, Vec<usize>> = HashMap::new();
+        for (i, (_, fork)) in blocks.iter().enumerate() {
+            by_fork.entry(*fork).or_default().push(i);
+        }
 
-            if chunk_bodies.len() != chunk.len() {
-                return Err(Error::InvalidPayloadBodiesResponse {
-                    expected: chunk.len(),
-                    received: chunk_bodies.len(),
-                });
+        let max_count = self
+            .execution_layer
+            .max_payload_bodies_per_request()
+            .await
+            .map_err(|e| Error::BlocksByHashFailure(Box::new(e)))?;
+
+        for (fork, indices) in by_fork {
+            for chunk in indices.chunks(max_count) {
+                let hashes = chunk.iter().map(|&i| blocks[i].0).collect::<Vec<_>>();
+                let chunk_bodies = self
+                    .execution_layer
+                    .get_payload_bodies_by_hash_v1(fork, hashes)
+                    .await
+                    .map_err(|e| Error::BlocksByHashFailure(Box::new(e)))?;
+
+                if chunk_bodies.len() != chunk.len() {
+                    return Err(Error::InvalidPayloadBodiesResponse {
+                        expected: chunk.len(),
+                        received: chunk_bodies.len(),
+                    });
+                }
+                for (&i, body) in chunk.iter().zip(chunk_bodies) {
+                    bodies[i] = body;
+                }
             }
-            bodies.extend(chunk_bodies);
         }
         Ok(bodies)
     }
@@ -278,17 +313,19 @@ impl<T: BeaconChainTypes> BeaconBlockStreamer<T> {
             }
         };
 
-        let block_hashes = loaded_blocks
+        let blocks = loaded_blocks
             .iter()
             .filter_map(|(_, loaded)| match loaded {
-                LoadedBlock::NeedsPayload(block_parts) => Some(block_parts.block_hash()),
+                LoadedBlock::NeedsPayload(block_parts) => {
+                    Some((block_parts.block_hash(), block_parts.fork()))
+                }
                 LoadedBlock::Complete(_) => None,
             })
             .collect::<Vec<_>>();
         // On failure, complete blocks are still sent and each block requiring a payload gets
         // the (shared) error as its result, so that the stream terminates with an error
         // rather than looking like a complete response.
-        let mut bodies = match self.fetch_payload_bodies(block_hashes).await {
+        let mut bodies = match self.fetch_payload_bodies(blocks).await {
             Ok(bodies) => Ok(bodies.into_iter()),
             Err(e) => {
                 let block_result: Arc<BlockResult<T::EthSpec>> = Arc::new(Err(e.into()));

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -2337,8 +2337,9 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             .as_ref()
             .ok_or(Error::ExecutionLayerMissing)?;
 
+        let fork = self.spec.fork_name_at_slot::<T::EthSpec>(request_slot);
         let inclusion_list_transactions = execution_layer
-            .get_inclusion_list_v1()
+            .get_inclusion_list_v1(fork)
             .await
             .map_err(|e| Error::ExecutionLayerGetInclusionListFailed(Box::new(e)))?;
 
@@ -7343,6 +7344,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 return Ok(());
             };
 
+        let prepare_fork = self.spec.fork_name_at_slot::<T::EthSpec>(current_slot + 1);
+
         let forkchoice_updated_response = execution_layer
             .notify_forkchoice_updated(
                 head_hash,
@@ -7352,6 +7355,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 head_block_root,
                 head_payload_status,
                 custody_columns,
+                prepare_fork,
             )
             .await
             .map_err(Error::ExecutionForkChoiceUpdateFailed);

beacon_node/beacon_chain/src/fetch_blobs/fetch_blobs_beacon_adapter.rs

@@ -161,7 +161,7 @@ impl<T: BeaconChainTypes> FetchBlobsBeaconAdapter<T> {
             .get_engine_capabilities(None)
             .await
             .map_err(FetchEngineBlobError::RequestFailed)
-            .map(|caps| caps.get_blobs_v3)
+            .map(|caps| caps.get_blobs_v3())
     }
 
     pub(crate) async fn supports_get_blobs_v4(&self) -> Result<bool, FetchEngineBlobError> {
@@ -175,6 +175,6 @@ impl<T: BeaconChainTypes> FetchBlobsBeaconAdapter<T> {
             .get_engine_capabilities(None)
             .await
             .map_err(FetchEngineBlobError::RequestFailed)
-            .map(|caps| caps.get_blobs_v4)
+            .map(|caps| caps.get_blobs_v4())
     }
 }

beacon_node/beacon_chain/src/graffiti_calculator.rs

@@ -253,8 +253,7 @@ mod tests {
     use crate::test_utils::{BeaconChainHarness, EphemeralHarnessType, test_spec};
     use bls::Keypair;
     use eth2::types::GraffitiPolicy;
-    use execution_layer::EngineCapabilities;
-    use execution_layer::test_utils::{DEFAULT_CLIENT_VERSION, DEFAULT_ENGINE_CAPABILITIES};
+    use execution_layer::test_utils::DEFAULT_CLIENT_VERSION;
     use std::sync::Arc;
     use std::sync::LazyLock;
     use std::time::Duration;
@@ -290,12 +289,7 @@ mod tests {
         let harness = get_harness(VALIDATOR_COUNT, spec, None);
         // modify execution engine so it doesn't support engine_getClientVersionV1 method
         let mock_execution_layer = harness.mock_execution_layer.as_ref().unwrap();
-        mock_execution_layer
-            .server
-            .set_engine_capabilities(EngineCapabilities {
-                get_client_version_v1: false,
-                ..DEFAULT_ENGINE_CAPABILITIES
-            });
+        mock_execution_layer.server.disable_client_version();
         // refresh capabilities cache
         harness
             .chain
@@ -334,12 +328,7 @@ mod tests {
         let spec = Arc::new(test_spec::<Spec>());
         let harness = get_harness(VALIDATOR_COUNT, spec, None);
         let mock_execution_layer = harness.mock_execution_layer.as_ref().unwrap();
-        mock_execution_layer
-            .server
-            .set_engine_capabilities(EngineCapabilities {
-                get_client_version_v1: false,
-                ..DEFAULT_ENGINE_CAPABILITIES
-            });
+        mock_execution_layer.server.disable_client_version();
         harness
             .chain
             .execution_layer

beacon_node/beacon_chain/src/payload_envelope_streamer/beacon_chain_adapter.rs

@@ -5,7 +5,8 @@ use execution_layer::ExecutionPayloadBodyV2;
 use mockall::automock;
 use task_executor::TaskExecutor;
 use types::{
-    ExecutionBlockHash, ExecutionPayloadBody, Hash256, SignedExecutionPayloadEnvelopeSummary, Slot,
+    ExecutionBlockHash, ExecutionPayloadBody, ForkName, Hash256,
+    SignedExecutionPayloadEnvelopeSummary, Slot,
 };
 
 use super::Error;
@@ -43,6 +44,7 @@ impl<T: BeaconChainTypes> EnvelopeStreamerBeaconAdapter<T> {
 
     pub(crate) async fn get_payload_bodies_by_hash_v2(
         &self,
+        fork: ForkName,
         block_hashes: Vec<ExecutionBlockHash>,
     ) -> Result<Vec<Option<ExecutionPayloadBodyV2<T::EthSpec>>>, BeaconChainError> {
         let execution_layer = self
@@ -52,11 +54,28 @@ impl<T: BeaconChainTypes> EnvelopeStreamerBeaconAdapter<T> {
             .ok_or(BeaconChainError::ExecutionLayerMissing)?;
 
         execution_layer
-            .get_payload_bodies_by_hash_v2(block_hashes)
+            .get_payload_bodies_by_hash_v2(fork, block_hashes)
             .await
             .map_err(|error| Error::PayloadBodiesByHashV2Failure(Box::new(error)).into())
     }
 
+    pub(crate) async fn max_payload_bodies_per_request(&self) -> Result<usize, BeaconChainError> {
+        let execution_layer = self
+            .chain
+            .execution_layer
+            .as_ref()
+            .ok_or(BeaconChainError::ExecutionLayerMissing)?;
+
+        execution_layer
+            .max_payload_bodies_per_request()
+            .await
+            .map_err(|error| Error::PayloadBodiesByHashV2Failure(Box::new(error)).into())
+    }
+
+    pub(crate) fn fork_name_at_slot(&self, slot: Slot) -> ForkName {
+        self.chain.spec.fork_name_at_slot::<T::EthSpec>(slot)
+    }
+
     pub(crate) fn get_split(&self) -> (Slot, Hash256) {
         let split = self.chain.store.get_split_info();
         (split.slot, split.block_root)