address review comments, inline safe_read_http_response (name implied

general but mux-specific)

Author: JasonVranek

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::read_chunked_body_with_max,
 };
 
 #[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);
@@ -237,7 +238,16 @@ impl MuxKeysLoader {
                 }
                 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 status = response.status();
+                let pubkey_bytes = read_chunked_body_with_max(response, MUXER_HTTP_MAX_LENGTH)
+                    .await
+                    .wrap_err("Failed to read response body")?;
+                if !status.is_success() {
+                    bail!(
+                        "Request failed with status: {status}, body: {}",
+                        String::from_utf8_lossy(&pubkey_bytes)
+                    );
+                }
                 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},
-    wire::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

@@ -1,13 +1,14 @@
 use std::time::Duration;
 
 use alloy::primitives::U256;
-use eyre::Context;
+use eyre::{Context, bail};
 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::read_chunked_body_with_max,
 };
 
 pub async fn request_ssv_pubkeys_from_ssv_node(
@@ -28,7 +29,16 @@ pub async fn request_ssv_pubkeys_from_ssv_node(
     })?;
 
     // Parse the response as JSON
-    let body_bytes = safe_read_http_response(response).await?;
+    let status = response.status();
+    let body_bytes = read_chunked_body_with_max(response, MUXER_HTTP_MAX_LENGTH)
+        .await
+        .wrap_err("Failed to read response body")?;
+    if !status.is_success() {
+        bail!(
+            "Request failed with status: {status}, body: {}",
+            String::from_utf8_lossy(&body_bytes)
+        );
+    }
     serde_json::from_slice::<SSVNodeResponse>(&body_bytes).wrap_err("failed to parse SSV response")
 }
 
@@ -46,7 +56,16 @@ pub async fn request_ssv_pubkeys_from_public_api(
     })?;
 
     // Parse the response as JSON
-    let body_bytes = safe_read_http_response(response).await?;
+    let status = response.status();
+    let body_bytes = read_chunked_body_with_max(response, MUXER_HTTP_MAX_LENGTH)
+        .await
+        .wrap_err("Failed to read response body")?;
+    if !status.is_success() {
+        bail!(
+            "Request failed with status: {status}, body: {}",
+            String::from_utf8_lossy(&body_bytes)
+        );
+    }
     serde_json::from_slice::<SSVPublicResponse>(&body_bytes)
         .wrap_err("failed to parse SSV response")
 }

crates/common/src/ssz.rs

@@ -1,22 +1,13 @@
 use alloy::primitives::U256;
 use lh_bls::Signature;
 use lh_types::ForkName;
-use ssz::{BYTES_PER_LENGTH_OFFSET, Decode, Encode};
+use ssz::BYTES_PER_LENGTH_OFFSET;
 
 use crate::pbs::{
     BuilderBidFulu, ExecutionPayloadHeaderFulu, ExecutionRequests, KzgCommitments,
     error::SszValueError,
 };
 
-/// Test that SSZ encoding and decoding round-trips, returning the decoded
-/// struct.
-pub fn test_encode_decode_ssz<T: Encode + Decode>(d: &[u8]) -> T {
-    let decoded = T::from_ssz_bytes(d).expect("deserialize");
-    let encoded = T::as_ssz_bytes(&decoded);
-    assert_eq!(encoded, d);
-    decoded
-}
-
 // Get the offset of the message in a SignedBuilderBid SSZ structure
 fn get_ssz_value_offset_for_fork(fork: ForkName) -> Result<usize, SszValueError> {
     match fork {

crates/common/src/wire.rs

@@ -23,10 +23,8 @@ pub const CONSENSUS_VERSION_HEADER: &str = "Eth-Consensus-Version";
 
 #[derive(Debug, Error)]
 pub enum ResponseReadError {
-    #[error(
-        "response size exceeds max size; max: {max}, content_length: {content_length}, raw: {raw}"
-    )]
-    PayloadTooLarge { max: usize, content_length: usize, raw: String },
+    #[error("response size exceeds max size; max: {max}, content_length: {content_length}")]
+    PayloadTooLarge { max: usize, content_length: usize },
 
     #[error("error reading response stream: {0}")]
     ReqwestError(#[from] reqwest::Error),
@@ -74,7 +72,6 @@ pub async fn read_chunked_body_with_max(
         return Err(ResponseReadError::PayloadTooLarge {
             max: max_size,
             content_length: length as usize,
-            raw: String::new(), // raw content is not available here
         });
     }
 
@@ -89,7 +86,6 @@ pub async fn read_chunked_body_with_max(
             return Err(ResponseReadError::PayloadTooLarge {
                 max: max_size,
                 content_length: content_length.unwrap_or(0) as usize,
-                raw: String::from_utf8_lossy(&response_bytes).into_owned(),
             });
         }
 

tests/tests/pbs_mux.rs

@@ -86,10 +86,9 @@ async fn test_ssv_network_fetch_big_data() -> Result<()> {
             panic!("Expected an error due to big content length, but got a successful response")
         }
         Err(e) => match e.downcast_ref::<ResponseReadError>() {
-            Some(ResponseReadError::PayloadTooLarge { max, content_length, raw }) => {
+            Some(ResponseReadError::PayloadTooLarge { max, content_length }) => {
                 assert_eq!(*max, MUXER_HTTP_MAX_LENGTH);
                 assert!(*content_length > MUXER_HTTP_MAX_LENGTH);
-                assert!(raw.is_empty());
             }
             _ => panic!("Expected PayloadTooLarge error, got: {}", e),
         },
@@ -147,10 +146,9 @@ async fn test_ssv_network_fetch_big_data_without_content_length() -> Result<()>
             panic!("Expected an error due to excessive data, but got a successful response")
         }
         Err(e) => match e.downcast_ref::<ResponseReadError>() {
-            Some(ResponseReadError::PayloadTooLarge { max, content_length, raw }) => {
+            Some(ResponseReadError::PayloadTooLarge { max, content_length }) => {
                 assert_eq!(*max, MUXER_HTTP_MAX_LENGTH);
                 assert_eq!(*content_length, 0);
-                assert!(!raw.is_empty());
             }
             _ => panic!("Expected PayloadTooLarge error, got: {}", e),
         },