Add SSZ types, content negotiation, and encoding helpers

Cargo.toml

@@ -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" }

crates/common/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/src/config/mux.rs

@@ -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/utils.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/interop/ssv/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/lib.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/pbs/error.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/mod.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/types/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}`