Implement builder ssz flow

Cargo.toml

@@ -16,7 +16,7 @@ members = [
 async-trait = "0.1"
 axum = { version = "0.7", features = ["ws"] }
 bytes = "1.6"
-eth2 = { git = "https://github.com/sigp/lighthouse.git", rev = "c33307d70287fd3b7a70785f89dadcb737214903" }
+eth2 = { git = "https://github.com/eserilev/lighthouse.git", rev = "098d5b5270df3cc5564d573be42965b1e2f623db" }
 ethereum_serde_utils = "0.7"
 ethereum_ssz = "0.7"
 ethereum_ssz_derive = "0.7"

builder-client/Cargo.toml

@@ -9,3 +9,5 @@ ethereum-apis-common = { path = "../common" }
 reqwest.workspace = true
 serde.workspace = true
 serde_json.workspace = true
+ethereum_ssz.workspace = true
+axum.workspace = true

builder-client/src/lib.rs

@@ -1,14 +1,21 @@
+use axum::http::HeaderMap;
+use axum::http::HeaderValue;
 pub use builder_api_types::*;
 pub use builder_bid::SignedBuilderBid;
+use ethereum_apis_common::ContentType;
 pub use ethereum_apis_common::ErrorResponse;
+use reqwest::header::CONTENT_TYPE;
 use reqwest::Client;
 use reqwest::Url;
 use serde::de::DeserializeOwned;
+use ssz::DecodeError;
+use ssz::Encode;
 
 #[derive(Debug)]
 pub enum Error {
     Reqwest(reqwest::Error),
     InvalidJson(serde_json::Error, String),
+    InvalidSsz(DecodeError),
     ServerMessage(ErrorResponse),
     StatusCode(reqwest::StatusCode),
     InvalidUrl(Url),
@@ -34,6 +41,34 @@ impl BuilderClient {
         }
     }
 
+    async fn build_response_with_headers<T>(
+        &self,
+        response: reqwest::Response,
+        content_type: ContentType,
+        fork_name: ForkName,
+    ) -> Result<T, Error>
+    where
+        T: DeserializeOwned + ForkVersionDecode,
+    {
+        let status = response.status();
+        let text = response.text().await?;
+
+        if status.is_success() {
+            match content_type {
+                ContentType::Json => {
+                    serde_json::from_str(&text).map_err(|e| Error::InvalidJson(e, text))
+                }
+                ContentType::Ssz => {
+                    T::from_ssz_bytes_by_fork(text.as_bytes(), fork_name).map_err(Error::InvalidSsz)
+                }
+            }
+        } else {
+            Err(Error::ServerMessage(
+                serde_json::from_str(&text).map_err(|e| Error::InvalidJson(e, text))?,
+            ))
+        }
+    }
+
     async fn build_response<T>(&self, response: reqwest::Response) -> Result<T, Error>
     where
         T: DeserializeOwned,
@@ -67,22 +102,50 @@ impl BuilderClient {
     pub async fn submit_blinded_block<E: EthSpec>(
         &self,
         block: &SignedBlindedBeaconBlock<E>,
+        content_type: ContentType,
+        fork_name: ForkName,
     ) -> Result<ExecutionPayload<E>, Error> {
         let mut url = self.base_url.clone();
         url.path_segments_mut()
             .map_err(|_| Error::InvalidUrl(self.base_url.clone()))?
             .extend(&["eth", "v1", "builder", "blinded_blocks"]);
 
-        let response = self.client.post(url).json(block).send().await?;
-
-        self.build_response(response).await
+        let mut headers = HeaderMap::new();
+        headers.insert(
+            CONTENT_TYPE,
+            HeaderValue::from_str(&content_type.to_string()).unwrap(),
+        );
+
+        let response = match content_type {
+            ContentType::Json => {
+                self.client
+                    .post(url)
+                    .headers(headers)
+                    .json(block)
+                    .send()
+                    .await?
+            }
+            ContentType::Ssz => {
+                self.client
+                    .post(url)
+                    .headers(headers)
+                    .body(block.as_ssz_bytes())
+                    .send()
+                    .await?
+            }
+        };
+
+        self.build_response_with_headers(response, content_type, fork_name)
+            .await
     }
 
     pub async fn get_header<E: EthSpec>(
         &self,
         slot: Slot,
         parent_hash: ExecutionBlockHash,
         pubkey: &PublicKeyBytes,
+        content_type: ContentType,
+        fork_name: ForkName,
     ) -> Result<SignedBuilderBid<E>, Error> {
         let mut url = self.base_url.clone();
         url.path_segments_mut()
@@ -97,9 +160,16 @@ impl BuilderClient {
                 &pubkey.to_string(),
             ]);
 
-        let response = self.client.get(url).send().await?;
+        let mut headers = HeaderMap::new();
+        headers.insert(
+            CONTENT_TYPE,
+            HeaderValue::from_str(&content_type.to_string()).unwrap(),
+        );
 
-        self.build_response(response).await
+        let response = self.client.get(url).headers(headers).send().await?;
+
+        self.build_response_with_headers(response, content_type, fork_name)
+            .await
     }
 
     pub async fn get_status(&self) -> Result<(), Error> {

builder-server/src/server.rs

@@ -7,11 +7,13 @@ use axum::{
     Json, Router,
 };
 use builder_api_types::{
-    eth_spec::EthSpec, fork_versioned_response::EmptyMetadata, ExecutionBlockHash,
-    ForkVersionedResponse, FullPayloadContents, PublicKeyBytes, SignedBlindedBeaconBlock,
+    eth_spec::EthSpec, ExecutionBlockHash, PublicKeyBytes, SignedBlindedBeaconBlock,
     SignedValidatorRegistrationData, Slot,
 };
-use ethereum_apis_common::build_response;
+use ethereum_apis_common::{
+    build_response, build_response_with_headers, ContentType, JsonOrSszWithFork,
+};
+use http::{header::CONTENT_TYPE, HeaderMap};
 
 use crate::builder::Builder;
 
@@ -52,39 +54,33 @@ where
 }
 
 async fn submit_blinded_block<I, A, E>(
+    headers: HeaderMap,
     State(api_impl): State<I>,
-    Json(block): Json<SignedBlindedBeaconBlock<E>>,
+    JsonOrSszWithFork(block): JsonOrSszWithFork<SignedBlindedBeaconBlock<E>>,
 ) -> Result<Response<Body>, StatusCode>
 where
     E: EthSpec,
     I: AsRef<A> + Send + Sync,
     A: Builder<E>,
 {
-    let res = api_impl
-        .as_ref()
-        .submit_blinded_block(block)
-        .await
-        .map(|payload| {
-            let fork_name = match &payload {
-                FullPayloadContents::Payload(payload) => payload.fork_name(),
-                FullPayloadContents::PayloadAndBlobs(payload_and_blobs) => {
-                    payload_and_blobs.execution_payload.fork_name()
-                }
-            };
-            ForkVersionedResponse {
-                version: Some(fork_name),
-                metadata: EmptyMetadata {},
-                data: payload,
-            }
-        });
-    build_response(res).await
+    let content_type_header = headers.get(CONTENT_TYPE);
+    let content_type = content_type_header.and_then(|value| value.to_str().ok());
+    let content_type = match content_type {
+        Some("application/octet-stream") => ContentType::Ssz,
+        _ => ContentType::Json,
+    };
+    let slot = block.slot();
+    let res = api_impl.as_ref().submit_blinded_block(block).await;
+
+    build_response_with_headers(res, content_type, api_impl.as_ref().fork_name_at_slot(slot)).await
 }
 
 async fn get_status() -> StatusCode {
     StatusCode::OK
 }
 
 async fn get_header<I, A, E>(
+    headers: HeaderMap,
     State(api_impl): State<I>,
     Path((slot, parent_hash, pubkey)): Path<(Slot, ExecutionBlockHash, PublicKeyBytes)>,
 ) -> Result<Response<Body>, StatusCode>
@@ -93,14 +89,15 @@ where
     I: AsRef<A> + Send + Sync,
     A: Builder<E>,
 {
+    let content_type_header = headers.get(CONTENT_TYPE);
+    let content_type = content_type_header.and_then(|value| value.to_str().ok());
+    let content_type = match content_type {
+        Some("application/octet-stream") => ContentType::Ssz,
+        _ => ContentType::Json,
+    };
     let res = api_impl
         .as_ref()
         .get_header(slot, parent_hash, pubkey)
-        .await
-        .map(|signed_bid| ForkVersionedResponse {
-            version: Some(api_impl.as_ref().fork_name_at_slot(slot)),
-            metadata: EmptyMetadata {},
-            data: signed_bid,
-        });
-    build_response(res).await
+        .await;
+    build_response_with_headers(res, content_type, api_impl.as_ref().fork_name_at_slot(slot)).await
 }