Add a REST-SSZ Engine API transport

Cargo.toml

@@ -214,7 +214,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"] }
+reqwest = { version = "0.12", default-features = false, features = ["blocking", "json", "stream", "rustls-tls", "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

@@ -6979,6 +6979,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,
@@ -6988,6 +6990,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

@@ -256,8 +256,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;
@@ -293,12 +292,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

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)

beacon_node/beacon_chain/src/payload_envelope_streamer/mod.rs

@@ -2,6 +2,7 @@ mod beacon_chain_adapter;
 #[cfg(test)]
 mod tests;
 
+use std::collections::HashMap;
 use std::sync::Arc;
 
 #[cfg_attr(test, double)]
@@ -14,7 +15,7 @@ use tokio::sync::mpsc::{self, UnboundedSender};
 use tokio_stream::wrappers::UnboundedReceiverStream;
 use tracing::{debug, error, warn};
 use types::{
-    EthSpec, ExecutionBlockHash, ExecutionPayloadRef, ExecutionRequestsRef, Hash256,
+    EthSpec, ExecutionBlockHash, ExecutionPayloadRef, ExecutionRequestsRef, ForkName, Hash256,
     SignedExecutionPayloadEnvelope, SignedExecutionPayloadEnvelopeSummary, Slot,
 };
 
@@ -163,7 +164,10 @@ impl<T: BeaconChainTypes> PayloadEnvelopeStreamer<T> {
         let payload_requests = loaded_envelopes
             .iter()
             .filter_map(|(_, loaded)| match loaded {
-                LoadedEnvelope::NeedsPayload(summary) => Some(summary.block_hash()),
+                LoadedEnvelope::NeedsPayload(summary) => Some((
+                    summary.block_hash(),
+                    self.adapter.fork_name_at_slot(summary.slot()),
+                )),
                 LoadedEnvelope::Complete(_) => None,
             })
             .collect::<Vec<_>>();
@@ -196,28 +200,49 @@ impl<T: BeaconChainTypes> PayloadEnvelopeStreamer<T> {
             .collect())
     }
 
-    /// Fetch Gloas payload bodies using `engine_getPayloadBodiesByHashV2`.
+    /// Fetch payload bodies for `blocks` (hash + fork) using `engine_getPayloadBodiesByHashV2`
     ///
-    /// The returned vector has the same length and order as `block_hashes`. An unknown body is
-    /// represented by `None`.
+    /// The returned vector has the same length and order as `blocks`. An unknown body is
+    /// represented by `None`. Requests are grouped by fork so each REST-selected engine request
+    /// is single-fork.
     async fn fetch_payload_bodies(
         &self,
-        block_hashes: Vec<ExecutionBlockHash>,
+        blocks: Vec<(ExecutionBlockHash, ForkName)>,
     ) -> Result<Vec<Option<ExecutionPayloadBodyV2<T::EthSpec>>>, BeaconChainError> {
-        let mut payload_bodies = Vec::with_capacity(block_hashes.len());
-        for chunk in block_hashes.chunks(MAX_PAYLOAD_BODIES_PER_REQUEST) {
-            let chunk_payload_bodies = self
-                .adapter
-                .get_payload_bodies_by_hash_v2(chunk.to_vec())
-                .await?;
-            if chunk_payload_bodies.len() != chunk.len() {
-                return Err(Error::InvalidPayloadBodiesResponse {
-                    expected: chunk.len(),
-                    received: chunk_payload_bodies.len(),
+        if blocks.is_empty() {
+            return Ok(Vec::new());
+        }
+
+        // Result slots by original index, so the returned order matches `blocks`.
+        let mut payload_bodies: Vec<Option<ExecutionPayloadBodyV2<T::EthSpec>>> =
+            vec![None; blocks.len()];
+
+        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);
+        }
+
+        let max_count = self.adapter.max_payload_bodies_per_request().await?;
+
+        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_payload_bodies = self
+                    .adapter
+                    .get_payload_bodies_by_hash_v2(fork, hashes)
+                    .await?;
+
+                if chunk_payload_bodies.len() != chunk.len() {
+                    return Err(Error::InvalidPayloadBodiesResponse {
+                        expected: chunk.len(),
+                        received: chunk_payload_bodies.len(),
+                    }
+                    .into());
+                }
+                for (&i, body) in chunk.iter().zip(chunk_payload_bodies) {
+                    payload_bodies[i] = body;
                 }
-                .into());
             }
-            payload_bodies.extend(chunk_payload_bodies);
         }
         Ok(payload_bodies)
     }
@@ -272,9 +297,6 @@ pub fn launch_payload_envelope_stream<T: BeaconChainTypes>(
     PayloadEnvelopeStreamer::new(adapter, request_source).launch_stream(block_roots)
 }
 
-/// The Engine API only guarantees support for 32 hashes per payload-body request.
-const MAX_PAYLOAD_BODIES_PER_REQUEST: usize = 32;
-
 fn reconstruct_envelope_from_body<E: EthSpec>(
     summary: SignedExecutionPayloadEnvelopeSummary<E>,
     payload_body: ExecutionPayloadBodyV2<E>,