beacon: consensus: implement engine api getBlobs

beacon_node/beacon_chain/src/execution_payload.rs

@@ -24,7 +24,7 @@ use state_processing::per_block_processing::{
 use std::sync::Arc;
 use tokio::task::JoinHandle;
 use tree_hash::TreeHash;
-use types::*;
+use types::{*, execution_payload::BlobsBundle};
 
 pub type PreparePayloadResult<Payload> = Result<Payload, BlockProductionError>;
 pub type PreparePayloadHandle<Payload> = JoinHandle<Option<PreparePayloadResult<Payload>>>;
@@ -483,5 +483,13 @@ where
         .await
         .map_err(BlockProductionError::GetPayloadFailed)?;
 
+    /* 
+    TODO: fetch blob bundles from el engine for block building
+    let suggested_fee_recipient = execution_layer.get_suggested_fee_recipient(proposer_index).await;
+    let blobs = execution_layer.get_blob_bundles(parent_hash, timestamp, random, suggested_fee_recipient)
+    .await
+    .map_err(BlockProductionError::GetPayloadFailed)?;
+    */
+
     Ok(execution_payload)
 }

beacon_node/execution_layer/src/engine_api/http.rs

@@ -681,19 +681,6 @@ impl HttpJsonRpc {
         Ok(response.into())
     }
 
-    pub async fn get_full_payload<T: EthSpec>(
-        &self,
-        payload_id: PayloadId,
-    ) -> Result<FullPayload<T>, Error> {
-        let payload = self.get_payload_v1(payload_id).await;
-        let blobs = self.get_blobs_bundle_v1(payload_id).await;
-
-        Ok(FullPayload{
-            execution_payload: payload?,
-            blobs_bundle: blobs?.into(),
-        })
-    }
-
     pub async fn forkchoice_updated_v1(
         &self,
         forkchoice_state: ForkChoiceState,

beacon_node/execution_layer/src/lib.rs

@@ -20,6 +20,7 @@ use sensitive_url::SensitiveUrl;
 use serde::{Deserialize, Serialize};
 use slog::{crit, debug, error, info, trace, warn, Logger};
 use slot_clock::SlotClock;
+use types::execution_payload::BlobsBundle;
 use std::collections::HashMap;
 use std::future::Future;
 use std::io::Write;
@@ -759,6 +760,55 @@ impl<T: EthSpec> ExecutionLayer<T> {
         .await
     }
 
+    pub async fn get_blob_bundles(
+        &self,
+        parent_hash: ExecutionBlockHash,
+        timestamp: u64,
+        prev_randao: Hash256,
+        suggested_fee_recipient: Address,
+    ) -> Result<BlobsBundle<T>, Error> {
+        debug!(
+            self.log(),
+            "Issuing engine_getPayload";
+            "suggested_fee_recipient" => ?suggested_fee_recipient,
+            "prev_randao" => ?prev_randao,
+            "timestamp" => timestamp,
+            "parent_hash" => ?parent_hash,
+        );
+        self.engine()
+            .request(|engine| async move {
+                let payload_id = if let Some(id) = engine
+                    .get_payload_id(parent_hash, timestamp, prev_randao, suggested_fee_recipient)
+                    .await
+                {
+                    // The payload id has been cached for this engine.
+                    metrics::inc_counter_vec(
+                        &metrics::EXECUTION_LAYER_PRE_PREPARED_PAYLOAD_ID,
+                        &[metrics::HIT],
+                    );
+                    id
+                } else {             
+                    error!(
+                        self.log(),
+                        "Exec engine unable to produce blobs, did you call get_payload before?",
+                    );
+                    return Err(ApiError::PayloadIdUnavailable);                       
+                };
+
+                engine
+                    .api
+                    .get_blobs_bundle_v1::<T>(payload_id)
+                    .await
+                    .map(|bundle| {
+                        // TODO verify the blob bundle here?
+                        bundle.into()
+                    })
+            })
+            .await
+            .map_err(Box::new)
+            .map_err(Error::EngineError)
+    }
+
     async fn get_full_payload_with<Payload: ExecPayload<T>>(
         &self,
         parent_hash: ExecutionBlockHash,
@@ -835,10 +885,10 @@ impl<T: EthSpec> ExecutionLayer<T> {
 
                 engine
                     .api
-                    .get_full_payload::<T>(payload_id)
+                    .get_payload_v1::<T>(payload_id)
                     .await
                     .map(|full_payload| {
-                        if full_payload.execution_payload.fee_recipient != suggested_fee_recipient {
+                        if full_payload.fee_recipient != suggested_fee_recipient {
                             error!(
                                 self.log(),
                                 "Inconsistent fee recipient";
@@ -847,11 +897,11 @@ impl<T: EthSpec> ExecutionLayer<T> {
                                 indicate that fees are being diverted to another address. Please \
                                 ensure that the value of suggested_fee_recipient is set correctly and \
                                 that the Execution Engine is trusted.",
-                                "fee_recipient" => ?full_payload.execution_payload.fee_recipient,
+                                "fee_recipient" => ?full_payload.fee_recipient,
                                 "suggested_fee_recipient" => ?suggested_fee_recipient,
                             );
                         }
-                        if f(self, &full_payload.execution_payload).is_some() {
+                        if f(self, &full_payload).is_some() {
                             warn!(
                                 self.log(),
                                 "Duplicate payload cached, this might indicate redundant proposal \

consensus/types/src/beacon_block_body.rs

@@ -232,7 +232,7 @@ impl<E: EthSpec> From<BeaconBlockBodyMerge<E, FullPayload<E>>>
             deposits,
             voluntary_exits,
             sync_aggregate,
-            execution_payload: FullPayload { execution_payload, blobs_bundle },
+            execution_payload: FullPayload { execution_payload },
         } = body;
 
         (
@@ -272,7 +272,7 @@ for (
             deposits,
             voluntary_exits,
             sync_aggregate,
-            execution_payload: FullPayload { execution_payload, blobs_bundle },
+            execution_payload: FullPayload { execution_payload},
             blob_kzg_commitments,
         } = body;
 
@@ -324,7 +324,7 @@ impl<E: EthSpec> BeaconBlockBodyMerge<E, FullPayload<E>> {
             deposits,
             voluntary_exits,
             sync_aggregate,
-            execution_payload: FullPayload { execution_payload, blobs_bundle },
+            execution_payload: FullPayload { execution_payload },
         } = self;
 
         BeaconBlockBodyMerge {
@@ -356,7 +356,7 @@ impl<E: EthSpec> BeaconBlockBodyEip4844<E, FullPayload<E>> {
             deposits,
             voluntary_exits,
             sync_aggregate,
-            execution_payload: FullPayload { execution_payload, blobs_bundle },
+            execution_payload: FullPayload { execution_payload },
             blob_kzg_commitments,
         } = self;
 

consensus/types/src/payload.rs

@@ -230,8 +230,7 @@ impl<T: EthSpec> Encode for BlindedPayload<T> {
 #[derive(Default, Debug, Clone, Serialize, Deserialize)]
 #[serde(bound = "T: EthSpec")]
 pub struct FullPayload<T: EthSpec> {
-    pub execution_payload: ExecutionPayload<T>,
-    pub blobs_bundle: Option<BlobsBundle<T>>,
+    pub execution_payload: ExecutionPayload<T>
 }
 
 impl <T: EthSpec> TestRandom for FullPayload<T> {
@@ -255,8 +254,7 @@ impl <T: EthSpec> Hash for FullPayload<T> {
 impl<T: EthSpec> From<ExecutionPayload<T>> for FullPayload<T> {
     fn from(execution_payload: ExecutionPayload<T>) -> Self {
         Self { 
-            execution_payload,
-            blobs_bundle: None,
+            execution_payload
         }
     }
 }
@@ -294,8 +292,7 @@ impl<T: EthSpec> Decode for FullPayload<T> {
 
     fn from_ssz_bytes(bytes: &[u8]) -> Result<Self, DecodeError> {
         Ok(FullPayload {
-            execution_payload: Decode::from_ssz_bytes(bytes)?,
-            blobs_bundle: None,
+            execution_payload: Decode::from_ssz_bytes(bytes)?
         })
     }
 }

consensus/types/src/signed_beacon_block.rs

@@ -304,7 +304,7 @@ impl<E: EthSpec> SignedBeaconBlockMerge<E, BlindedPayload<E>> {
                     deposits,
                     voluntary_exits,
                     sync_aggregate,
-                    execution_payload: FullPayload { execution_payload: execution_payload, blobs_bundle: None },
+                    execution_payload: FullPayload { execution_payload },
                 },
             },
             signature,
@@ -357,7 +357,7 @@ impl<E: EthSpec> SignedBeaconBlockEip4844<E, BlindedPayload<E>> {
                     deposits,
                     voluntary_exits,
                     sync_aggregate,
-                    execution_payload: FullPayload { execution_payload: execution_payload, blobs_bundle: None  },
+                    execution_payload: FullPayload { execution_payload },
                     blob_kzg_commitments,
                 },
             },