Add SSZ types, content negotiation, and encoding helpers (#466)

Author: JasonVranek

Cargo.lock

@@ -49,9 +49,11 @@ ethereum_ssz_derive = "0.10"
 eyre = "0.6.12"
 futures = "0.3.30"
 headers = "0.4.0"
+headers-accept = "0.2.1"
 indexmap = "2.2.6"
 jsonwebtoken = { version = "9.3.1", default-features = false }
 lazy_static = "1.5.0"
+mediatype = "0.20.0"
 lh_eth2 = { package = "eth2", git = "https://github.com/sigp/lighthouse", tag = "v8.1.3", features = ["events"] }
 lh_eth2_keystore = { package = "eth2_keystore", git = "https://github.com/sigp/lighthouse", tag = "v8.1.3" }
 lh_bls = { package = "bls", git = "https://github.com/sigp/lighthouse", tag = "v8.1.3" }

Cargo.toml

@@ -26,12 +26,14 @@ ethereum_ssz.workspace = true
 ethereum_ssz_derive.workspace = true
 eyre.workspace = true
 futures.workspace = true
+headers-accept.workspace = true
 jsonwebtoken.workspace = true
 lazy_static.workspace = true
 lh_bls.workspace = true
 lh_eth2.workspace = true
 lh_eth2_keystore.workspace = true
 lh_types.workspace = true
+mediatype.workspace = true
 notify.workspace = true
 pbkdf2.workspace = true
 rand.workspace = true

crates/common/Cargo.toml

@@ -18,13 +18,14 @@ use serde::{Deserialize, Serialize};
 use tracing::{debug, info, warn};
 use url::Url;
 
-use super::{MUX_PATH_ENV, PbsConfig, RelayConfig, load_optional_env_var};
+use super::{MUX_PATH_ENV, MUXER_HTTP_MAX_LENGTH, PbsConfig, RelayConfig, load_optional_env_var};
 use crate::{
-    config::{remove_duplicate_keys, safe_read_http_response},
+    config::remove_duplicate_keys,
     interop::{lido::utils::*, ssv::utils::*},
     pbs::RelayClient,
     types::{BlsPublicKey, Chain},
     utils::default_bool,
+    wire::safe_read_http_response,
 };
 
 #[derive(Debug, Clone, Deserialize, Serialize)]
@@ -156,7 +157,7 @@ pub struct MuxConfig {
 
 impl MuxConfig {
     /// Returns the env, actual path, and internal path to use for the file
-    /// loader. In File mode, validates the mux file prior to returning.   
+    /// loader. In File mode, validates the mux file prior to returning.
     pub fn loader_env(&self) -> eyre::Result<Option<(String, String, String)>> {
         let Some(loader) = self.loader.as_ref() else {
             return Ok(None);
@@ -235,9 +236,10 @@ impl MuxKeysLoader {
                         "Mux keys URL {url} is insecure; consider using HTTPS if possible instead"
                     );
                 }
+                let url = url.as_str();
                 let client = reqwest::ClientBuilder::new().timeout(http_timeout).build()?;
                 let response = client.get(url).send().await?;
-                let pubkey_bytes = safe_read_http_response(response).await?;
+                let pubkey_bytes = safe_read_http_response(response, MUXER_HTTP_MAX_LENGTH).await?;
                 serde_json::from_slice(&pubkey_bytes)
                     .wrap_err("failed to fetch mux keys from HTTP endpoint")
             }

crates/common/src/config/mux.rs

@@ -7,9 +7,8 @@ use eyre::{Context, Result, bail};
 use serde::de::DeserializeOwned;
 
 use crate::{
-    config::{ADMIN_JWT_ENV, JWTS_ENV, MUXER_HTTP_MAX_LENGTH},
+    config::{ADMIN_JWT_ENV, JWTS_ENV},
     types::{BlsPublicKey, ModuleId},
-    utils::read_chunked_body_with_max,
 };
 
 pub fn load_env_var(env: &str) -> Result<String> {
@@ -42,32 +41,6 @@ pub fn load_jwt_secrets() -> Result<(String, HashMap<ModuleId, String>)> {
     decode_string_to_map(&jwt_secrets).map(|secrets| (admin_jwt, secrets))
 }
 
-/// Reads an HTTP response safely, erroring out if it failed or if the body is
-/// too large.
-pub async fn safe_read_http_response(response: reqwest::Response) -> Result<Vec<u8>> {
-    // Read the response to a buffer in chunks
-    let status_code = response.status();
-    match read_chunked_body_with_max(response, MUXER_HTTP_MAX_LENGTH).await {
-        Ok(response_bytes) => {
-            if status_code.is_success() {
-                return Ok(response_bytes);
-            }
-            bail!(
-                "Request failed with status: {status_code}, body: {}",
-                String::from_utf8_lossy(&response_bytes)
-            )
-        }
-        Err(e) => {
-            if status_code.is_success() {
-                return Err(e).wrap_err("Failed to read response body");
-            }
-            Err(e).wrap_err(format!(
-                "Request failed with status {status_code}, but decoding the response body failed"
-            ))
-        }
-    }
-}
-
 /// Removes duplicate entries from a vector of BlsPublicKey
 pub fn remove_duplicate_keys(keys: Vec<BlsPublicKey>) -> Vec<BlsPublicKey> {
     let mut unique_keys = Vec::new();

crates/common/src/config/utils.rs

@@ -6,8 +6,9 @@ use serde_json::json;
 use url::Url;
 
 use crate::{
-    config::safe_read_http_response,
+    config::MUXER_HTTP_MAX_LENGTH,
     interop::ssv::types::{SSVNodeResponse, SSVPublicResponse},
+    wire::safe_read_http_response,
 };
 
 pub async fn request_ssv_pubkeys_from_ssv_node(
@@ -28,7 +29,7 @@ pub async fn request_ssv_pubkeys_from_ssv_node(
     })?;
 
     // Parse the response as JSON
-    let body_bytes = safe_read_http_response(response).await?;
+    let body_bytes = safe_read_http_response(response, MUXER_HTTP_MAX_LENGTH).await?;
     serde_json::from_slice::<SSVNodeResponse>(&body_bytes).wrap_err("failed to parse SSV response")
 }
 
@@ -46,7 +47,7 @@ pub async fn request_ssv_pubkeys_from_public_api(
     })?;
 
     // Parse the response as JSON
-    let body_bytes = safe_read_http_response(response).await?;
+    let body_bytes = safe_read_http_response(response, MUXER_HTTP_MAX_LENGTH).await?;
     serde_json::from_slice::<SSVPublicResponse>(&body_bytes)
         .wrap_err("failed to parse SSV response")
 }

crates/common/src/interop/ssv/utils.rs

@@ -7,7 +7,9 @@ pub mod interop;
 pub mod pbs;
 pub mod signature;
 pub mod signer;
+pub mod ssz;
 pub mod types;
 pub mod utils;
+pub mod wire;
 
 pub const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(12);

crates/common/src/lib.rs

@@ -1,7 +1,8 @@
 use alloy::primitives::{B256, U256};
+use lh_types::ForkName;
 use thiserror::Error;
 
-use crate::{types::BlsPublicKeyBytes, utils::ResponseReadError};
+use crate::{types::BlsPublicKeyBytes, wire::ResponseReadError};
 
 #[derive(Debug, Error)]
 pub enum PbsError {
@@ -28,31 +29,37 @@ pub enum PbsError {
 
     #[error("tokio join error: {0}")]
     TokioJoinError(#[from] tokio::task::JoinError),
+
+    #[error("SSZ error: {0}")]
+    SszError(#[from] SszValueError),
 }
 
 impl PbsError {
     pub fn is_timeout(&self) -> bool {
         matches!(self, PbsError::Reqwest(err) if err.is_timeout())
     }
 
+    /// Extract the HTTP status code from relay-originated errors.
+    fn relay_status_code(&self) -> Option<u16> {
+        match self {
+            PbsError::RelayResponse { code, .. } => Some(*code),
+            PbsError::ReadResponse(ResponseReadError::NonSuccess { status_code, .. }) => {
+                Some(*status_code)
+            }
+            _ => None,
+        }
+    }
+
     /// Whether the error is retryable in requests to relays
     pub fn should_retry(&self) -> bool {
         match self {
-            PbsError::Reqwest(err) => {
-                // Retry on timeout or connection error
-                err.is_timeout() || err.is_connect()
-            }
-            PbsError::RelayResponse { code, .. } => match *code {
-                500..509 => true,   // Retry on server errors
-                400 | 429 => false, // Do not retry if rate limited or bad request
-                _ => false,
-            },
-            _ => false,
+            PbsError::Reqwest(err) => err.is_timeout() || err.is_connect(),
+            _ => matches!(self.relay_status_code(), Some(500..=508)),
         }
     }
 
     pub fn is_not_found(&self) -> bool {
-        matches!(&self, PbsError::RelayResponse { code: 404, .. })
+        self.relay_status_code() == Some(404)
     }
 }
 
@@ -107,3 +114,12 @@ pub enum ValidationError {
     #[error("unsupported fork")]
     UnsupportedFork,
 }
+
+#[derive(Debug, Error, PartialEq, Eq)]
+pub enum SszValueError {
+    #[error("invalid payload length: required {required} but payload was {actual}")]
+    InvalidPayloadLength { required: usize, actual: usize },
+
+    #[error("unsupported fork: {name}")]
+    UnsupportedFork { name: ForkName },
+}

crates/common/src/pbs/error.rs

@@ -6,5 +6,6 @@ mod types;
 
 pub use builder::*;
 pub use constants::*;
+pub use lh_types::ForkVersionDecode;
 pub use relay::*;
 pub use types::*;

crates/common/src/pbs/mod.rs

@@ -1,5 +1,5 @@
 use alloy::primitives::{B256, U256, b256};
-use lh_eth2::ForkVersionedResponse;
+pub use lh_eth2::ForkVersionedResponse;
 pub use lh_types::ForkName;
 use lh_types::{BlindedPayload, ExecPayload, MainnetEthSpec};
 use serde::{Deserialize, Serialize};
@@ -26,6 +26,10 @@ pub type PayloadAndBlobs = lh_eth2::types::ExecutionPayloadAndBlobs<MainnetEthSp
 pub type SubmitBlindedBlockResponse = ForkVersionedResponse<PayloadAndBlobs>;
 
 pub type ExecutionPayloadHeader = lh_types::ExecutionPayloadHeader<MainnetEthSpec>;
+pub type ExecutionPayloadHeaderBellatrix =
+    lh_types::ExecutionPayloadHeaderBellatrix<MainnetEthSpec>;
+pub type ExecutionPayloadHeaderCapella = lh_types::ExecutionPayloadHeaderCapella<MainnetEthSpec>;
+pub type ExecutionPayloadHeaderDeneb = lh_types::ExecutionPayloadHeaderDeneb<MainnetEthSpec>;
 pub type ExecutionPayloadHeaderElectra = lh_types::ExecutionPayloadHeaderElectra<MainnetEthSpec>;
 pub type ExecutionPayloadHeaderFulu = lh_types::ExecutionPayloadHeaderFulu<MainnetEthSpec>;
 pub type ExecutionPayloadHeaderRef<'a> = lh_types::ExecutionPayloadHeaderRef<'a, MainnetEthSpec>;
@@ -34,7 +38,11 @@ pub type ExecutionPayloadElectra = lh_types::ExecutionPayloadElectra<MainnetEthS
 pub type ExecutionPayloadFulu = lh_types::ExecutionPayloadFulu<MainnetEthSpec>;
 pub type SignedBuilderBid = lh_types::SignedBuilderBid<MainnetEthSpec>;
 pub type BuilderBid = lh_types::BuilderBid<MainnetEthSpec>;
+pub type BuilderBidBellatrix = lh_types::BuilderBidBellatrix<MainnetEthSpec>;
+pub type BuilderBidCapella = lh_types::BuilderBidCapella<MainnetEthSpec>;
+pub type BuilderBidDeneb = lh_types::BuilderBidDeneb<MainnetEthSpec>;
 pub type BuilderBidElectra = lh_types::BuilderBidElectra<MainnetEthSpec>;
+pub type BuilderBidFulu = lh_types::BuilderBidFulu<MainnetEthSpec>;
 
 /// Response object of GET
 /// `/eth/v1/builder/header/{slot}/{parent_hash}/{pubkey}`