Implement builder ssz flow (#8)

* Impl builder ssz flow

* Update cargo

* fix header

* Use lighthouse unstable and re-export only required crates

* add logging

* Debug

* More debug

* Remove all debugs

* Update lighthouse version

* Add ssz support to register_validators

* Update lighthouse deps to add ssz support to SignedValidatorRegistrationData

* Add tests for builder server endpoints

* Remove debugs

---------

Co-authored-by: Pawan Dhananjay <[email protected]>

Author: eserilev

Cargo.toml

@@ -16,21 +16,22 @@ members = [
 
 [workspace.dependencies]
 async-trait = "0.1"
-axum = { version = "0.7", features = ["ws"] }
+axum = { version = "0.8", features = ["ws"] }
 bytes = "1.6"
-eth2 = { git = "https://github.com/sigp/lighthouse.git", rev = "c33307d70287fd3b7a70785f89dadcb737214903" }
+eth2 = { git = "https://github.com/sigp/lighthouse.git", rev = "3bc5f1f2a58b1df9454884672c8100fd5f79ba8b" }
 ethereum_serde_utils = "0.7"
 ethereum_ssz = "0.7"
 ethereum_ssz_derive = "0.7"
 flate2 = "1.0"
 futures = "0.3.30"
-http = "1"
-reqwest = { version = "0.12.5", features = ["json"] }
+http = "1.2"
+mediatype = "0.19.13"
+reqwest = { version = "0.12.12", features = ["json"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = { version = "1", features = ["raw_value"] }
 superstruct = "0.8"
 tokio = { version = "1", default-features = false, features = ["signal", "rt-multi-thread", "macros"] }
 tokio-tungstenite = "0.24.0"
 tracing = { version = "0.1", features = ["attributes"] }
-types = { git = "https://github.com/sigp/lighthouse.git", rev = "c33307d70287fd3b7a70785f89dadcb737214903" }
+types = { git = "https://github.com/sigp/lighthouse.git", rev = "3bc5f1f2a58b1df9454884672c8100fd5f79ba8b" }
 rand = "0.8"

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 @@
-pub use builder_api_types::*;
+use axum::http::HeaderMap;
+use axum::http::HeaderValue;
+use builder_api_types::*;
 pub use builder_bid::SignedBuilderBid;
+use ethereum_apis_common::ContentType;
 pub use ethereum_apis_common::ErrorResponse;
+use reqwest::header::{ACCEPT, 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(
+            ACCEPT,
+            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/Cargo.toml

@@ -17,3 +17,6 @@ serde.workspace = true
 serde_json.workspace = true
 tokio.workspace = true
 tracing.workspace = true
+
+[dev-dependencies]
+tower = "0.5.2"