wrap calls to read_chunked_body_with_max with safe_read_http_response

JasonVranek · JasonVranek · commit 4664fdd31143 · 2026-05-23T08:49:08.000-07:00
diff --git a/crates/common/src/config/mux.rs b/crates/common/src/config/mux.rs
@@ -25,7 +25,7 @@ use crate::{
     pbs::RelayClient,
     types::{BlsPublicKey, Chain},
     utils::default_bool,
-    wire::read_chunked_body_with_max,
+    wire::safe_read_http_response,
 };
 
 #[derive(Debug, Clone, Deserialize, Serialize)]
@@ -236,18 +236,11 @@ 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 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)
-                    );
-                }
+                let pubkey_bytes =
+                    safe_read_http_response(response, MUXER_HTTP_MAX_LENGTH, url).await?;
                 serde_json::from_slice(&pubkey_bytes)
                     .wrap_err("failed to fetch mux keys from HTTP endpoint")
             }
diff --git a/crates/common/src/interop/ssv/utils.rs b/crates/common/src/interop/ssv/utils.rs
@@ -1,21 +1,22 @@
 use std::time::Duration;
 
 use alloy::primitives::U256;
-use eyre::{Context, bail};
+use eyre::Context;
 use serde_json::json;
 use url::Url;
 
 use crate::{
     config::MUXER_HTTP_MAX_LENGTH,
     interop::ssv::types::{SSVNodeResponse, SSVPublicResponse},
-    wire::read_chunked_body_with_max,
+    wire::safe_read_http_response,
 };
 
 pub async fn request_ssv_pubkeys_from_ssv_node(
     url: Url,
     node_operator_id: U256,
     http_timeout: Duration,
 ) -> eyre::Result<SSVNodeResponse> {
+    let url = url.as_str();
     let client = reqwest::ClientBuilder::new().timeout(http_timeout).build()?;
     let body = json!({
         "operators": [node_operator_id]
@@ -29,23 +30,15 @@ pub async fn request_ssv_pubkeys_from_ssv_node(
     })?;
 
     // Parse the response as JSON
-    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)
-        );
-    }
+    let body_bytes = safe_read_http_response(response, MUXER_HTTP_MAX_LENGTH, url).await?;
     serde_json::from_slice::<SSVNodeResponse>(&body_bytes).wrap_err("failed to parse SSV response")
 }
 
 pub async fn request_ssv_pubkeys_from_public_api(
     url: Url,
     http_timeout: Duration,
 ) -> eyre::Result<SSVPublicResponse> {
+    let url = url.as_str();
     let client = reqwest::ClientBuilder::new().timeout(http_timeout).build()?;
     let response = client.get(url).send().await.map_err(|e| {
         if e.is_timeout() {
@@ -56,16 +49,7 @@ pub async fn request_ssv_pubkeys_from_public_api(
     })?;
 
     // Parse the response as JSON
-    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)
-        );
-    }
+    let body_bytes = safe_read_http_response(response, MUXER_HTTP_MAX_LENGTH, url).await?;
     serde_json::from_slice::<SSVPublicResponse>(&body_bytes)
         .wrap_err("failed to parse SSV response")
 }
diff --git a/crates/common/src/pbs/error.rs b/crates/common/src/pbs/error.rs
@@ -39,24 +39,27 @@ impl PbsError {
         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)
     }
 }
 
diff --git a/crates/common/src/wire.rs b/crates/common/src/wire.rs
@@ -23,11 +23,18 @@ 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}")]
-    PayloadTooLarge { max: usize, content_length: usize },
+    #[error(
+        "response size exceeds max size; max: {max}, content_length: {content_length}, request_url: {request_url}, request_id: {request_id}"
+    )]
+    PayloadTooLarge { max: usize, content_length: usize, request_url: String, request_id: String },
 
     #[error("error reading response stream: {0}")]
     ReqwestError(#[from] reqwest::Error),
+
+    #[error(
+        "request failed with status: {status_code}, request_url: {request_url}, request_id: {request_id}, body: {error_msg}"
+    )]
+    NonSuccess { status_code: u16, error_msg: String, request_url: String, request_id: String },
 }
 
 #[cfg(feature = "testing-flags")]
@@ -51,6 +58,8 @@ fn should_ignore_content_length() -> bool {
 pub async fn read_chunked_body_with_max(
     res: Response,
     max_size: usize,
+    request_id: &str,
+    request_url: &str,
 ) -> Result<Vec<u8>, ResponseReadError> {
     // Get the content length from the response headers
     #[cfg(not(feature = "testing-flags"))]
@@ -72,6 +81,8 @@ pub async fn read_chunked_body_with_max(
         return Err(ResponseReadError::PayloadTooLarge {
             max: max_size,
             content_length: length as usize,
+            request_url: request_url.to_string(),
+            request_id: request_id.to_string(),
         });
     }
 
@@ -86,6 +97,8 @@ pub async fn read_chunked_body_with_max(
             return Err(ResponseReadError::PayloadTooLarge {
                 max: max_size,
                 content_length: content_length.unwrap_or(0) as usize,
+                request_url: request_url.to_string(),
+                request_id: request_id.to_string(),
             });
         }
 
@@ -95,6 +108,28 @@ pub async fn read_chunked_body_with_max(
     Ok(response_bytes)
 }
 
+/// Reads an HTTP response body with a size limit, erroring on non-success
+/// status or read failure.
+pub async fn safe_read_http_response(
+    response: reqwest::Response,
+    max_size: usize,
+    request_id: &str,
+) -> Result<Vec<u8>, ResponseReadError> {
+    let status_code = response.status();
+    let request_url = response.url().to_string();
+    let body = read_chunked_body_with_max(response, max_size, request_id, &request_url).await?;
+    if status_code.is_success() {
+        Ok(body)
+    } else {
+        Err(ResponseReadError::NonSuccess {
+            status_code: status_code.as_u16(),
+            error_msg: String::from_utf8_lossy(&body).into_owned(),
+            request_url: request_url.to_string(),
+            request_id: request_id.to_string(),
+        })
+    }
+}
+
 /// Returns the user agent from the request headers or an empty string if not
 /// present
 pub fn get_user_agent(req_headers: &HeaderMap) -> String {
diff --git a/crates/pbs/src/mev_boost/get_header.rs b/crates/pbs/src/mev_boost/get_header.rs
@@ -19,7 +19,7 @@ use cb_common::{
     signature::verify_signed_message,
     types::{BlsPublicKey, BlsPublicKeyBytes, BlsSignature, Chain},
     utils::{ms_into_slot, timestamp_of_slot_start_sec, utcnow_ms},
-    wire::{get_user_agent_with_version, read_chunked_body_with_max},
+    wire::{get_user_agent_with_version, safe_read_http_response},
 };
 use futures::future::join_all;
 use parking_lot::RwLock;
@@ -351,14 +351,9 @@ async fn send_one_get_header(
     let code = res.status();
     RELAY_STATUS_CODE.with_label_values(&[code.as_str(), GET_HEADER_ENDPOINT_TAG, &relay.id]).inc();
 
-    let response_bytes = read_chunked_body_with_max(res, MAX_SIZE_GET_HEADER_RESPONSE).await?;
+    let response_bytes =
+        safe_read_http_response(res, MAX_SIZE_GET_HEADER_RESPONSE, relay.id.as_ref()).await?;
     let header_size_bytes = response_bytes.len();
-    if !code.is_success() {
-        return Err(PbsError::RelayResponse {
-            error_msg: String::from_utf8_lossy(&response_bytes).into_owned(),
-            code: code.as_u16(),
-        });
-    };
     if code == StatusCode::NO_CONTENT {
         debug!(
             relay_id = relay.id.as_ref(),
diff --git a/crates/pbs/src/mev_boost/register_validator.rs b/crates/pbs/src/mev_boost/register_validator.rs
@@ -5,7 +5,7 @@ use axum::http::{HeaderMap, HeaderValue};
 use cb_common::{
     pbs::{HEADER_START_TIME_UNIX_MS, RelayClient, error::PbsError},
     utils::utcnow_ms,
-    wire::{get_user_agent_with_version, read_chunked_body_with_max},
+    wire::{get_user_agent_with_version, safe_read_http_response},
 };
 use eyre::bail;
 use futures::{
@@ -187,17 +187,9 @@ async fn send_register_validator(
         .with_label_values(&[code.as_str(), REGISTER_VALIDATOR_ENDPOINT_TAG, &relay.id])
         .inc();
 
-    if !code.is_success() {
-        let response_bytes = read_chunked_body_with_max(res, MAX_SIZE_DEFAULT).await?;
-        let err = PbsError::RelayResponse {
-            error_msg: String::from_utf8_lossy(&response_bytes).into_owned(),
-            code: code.as_u16(),
-        };
-
-        // error here since we check if any success above
-        error!(relay_id = relay.id.as_ref(), retry, %err, "failed registration");
-        return Err(err);
-    };
+    safe_read_http_response(res, MAX_SIZE_DEFAULT, relay.id.as_ref()).await.inspect_err(|e| {
+        error!(relay_id = relay.id.as_ref(), retry, %e, "failed registration");
+    })?;
 
     debug!(
         relay_id = relay.id.as_ref(),
diff --git a/crates/pbs/src/mev_boost/status.rs b/crates/pbs/src/mev_boost/status.rs
@@ -3,7 +3,7 @@ use std::time::{Duration, Instant};
 use axum::http::HeaderMap;
 use cb_common::{
     pbs::{RelayClient, error::PbsError},
-    wire::{get_user_agent_with_version, read_chunked_body_with_max},
+    wire::{get_user_agent_with_version, safe_read_http_response},
 };
 use futures::future::select_ok;
 use reqwest::header::USER_AGENT;
@@ -73,16 +73,9 @@ async fn send_relay_check(relay: &RelayClient, headers: HeaderMap) -> Result<(),
     let code = res.status();
     RELAY_STATUS_CODE.with_label_values(&[code.as_str(), STATUS_ENDPOINT_TAG, &relay.id]).inc();
 
-    if !code.is_success() {
-        let response_bytes = read_chunked_body_with_max(res, MAX_SIZE_DEFAULT).await?;
-        let err = PbsError::RelayResponse {
-            error_msg: String::from_utf8_lossy(&response_bytes).into_owned(),
-            code: code.as_u16(),
-        };
-
-        error!(relay_id = relay.id.as_ref(),%err, "status failed");
-        return Err(err);
-    };
+    safe_read_http_response(res, MAX_SIZE_DEFAULT, relay.id.as_ref()).await.inspect_err(|e| {
+        error!(relay_id = relay.id.as_ref(), %e, "status failed");
+    })?;
 
     debug!(relay_id = relay.id.as_ref(),?code, latency = ?request_latency, "status passed");
 
diff --git a/crates/pbs/src/mev_boost/submit_block.rs b/crates/pbs/src/mev_boost/submit_block.rs
@@ -14,7 +14,7 @@ use cb_common::{
         error::{PbsError, ValidationError},
     },
     utils::utcnow_ms,
-    wire::{get_user_agent_with_version, read_chunked_body_with_max},
+    wire::{get_user_agent_with_version, safe_read_http_response},
 };
 use futures::{FutureExt, future::select_ok};
 use reqwest::header::USER_AGENT;
@@ -201,17 +201,11 @@ async fn send_submit_block(
         .with_label_values(&[code.as_str(), SUBMIT_BLINDED_BLOCK_ENDPOINT_TAG, &relay.id])
         .inc();
 
-    let response_bytes = read_chunked_body_with_max(res, MAX_SIZE_SUBMIT_BLOCK_RESPONSE).await?;
-    if !code.is_success() {
-        let err = PbsError::RelayResponse {
-            error_msg: String::from_utf8_lossy(&response_bytes).into_owned(),
-            code: code.as_u16(),
-        };
-
-        // we requested the payload from all relays, but some may have not received it
-        warn!(relay_id = relay.id.as_ref(), retry, %err, "failed to get payload (this might be ok if other relays have it)");
-        return Err(err);
-    };
+    let response_bytes = safe_read_http_response(res, MAX_SIZE_SUBMIT_BLOCK_RESPONSE, relay.id.as_ref())
+        .await
+        .inspect_err(|e| {
+            warn!(relay_id = relay.id.as_ref(), retry, %e, "failed to get payload (this might be ok if other relays have it)");
+        })?;
 
     if api_version != &BuilderApiVersion::V1 {
         // v2 response is going to be empty, so just break here
diff --git a/tests/tests/pbs_mux.rs b/tests/tests/pbs_mux.rs
@@ -78,17 +78,25 @@ async fn test_ssv_network_fetch_big_data() -> Result<()> {
     let server_handle =
         cb_tests::mock_ssv_public::create_mock_public_ssv_server(port, None).await?;
     let url = Url::parse(&format!("http://localhost:{port}/big_data")).unwrap();
-    let response = request_ssv_pubkeys_from_public_api(url, Duration::from_secs(120)).await;
+    let response = request_ssv_pubkeys_from_public_api(url.clone(), Duration::from_secs(120)).await;
 
     // The response should fail due to content length being too big
     match response {
         Ok(_) => {
             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 }) => {
+            Some(ResponseReadError::PayloadTooLarge {
+                max,
+                content_length,
+                request_url,
+                request_id,
+            }) => {
                 assert_eq!(*max, MUXER_HTTP_MAX_LENGTH);
                 assert!(*content_length > MUXER_HTTP_MAX_LENGTH);
+                assert_eq!(url.as_str(), request_url.as_str());
+                // url used as request id
+                assert_eq!(request_id.as_str(), request_url.as_str());
             }
             _ => panic!("Expected PayloadTooLarge error, got: {}", e),
         },
@@ -138,17 +146,25 @@ async fn test_ssv_network_fetch_big_data_without_content_length() -> Result<()>
     set_ignore_content_length(true);
     let server_handle = create_mock_public_ssv_server(port, None).await?;
     let url = Url::parse(&format!("http://localhost:{port}/big_data")).unwrap();
-    let response = request_ssv_pubkeys_from_public_api(url, Duration::from_secs(120)).await;
+    let response = request_ssv_pubkeys_from_public_api(url.clone(), Duration::from_secs(120)).await;
 
     // The response should fail due to the body being too big
     match response {
         Ok(_) => {
             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 }) => {
+            Some(ResponseReadError::PayloadTooLarge {
+                max,
+                content_length,
+                request_url,
+                request_id,
+            }) => {
                 assert_eq!(*max, MUXER_HTTP_MAX_LENGTH);
                 assert_eq!(*content_length, 0);
+                assert_eq!(url.as_str(), request_url.as_str());
+                // url used as request id
+                assert_eq!(request_id.as_str(), request_url.as_str());
             }
             _ => panic!("Expected PayloadTooLarge error, got: {}", e),
         },
