Merge branch 'lt/get-payload-v2-on-all' into add-ssz-to-pbs

jclapis · jclapis · commit 06a30929e72b · 2025-10-27T09:18:38.000-04:00
diff --git a/crates/common/src/config/mux.rs b/crates/common/src/config/mux.rs
@@ -369,7 +369,7 @@ async fn fetch_lido_registry_keys(
 }
 
 async fn fetch_ssv_pubkeys(
-    api_url: Url,
+    mut api_url: Url,
     chain: Chain,
     node_operator_id: U256,
     http_timeout: Duration,
@@ -386,6 +386,15 @@ async fn fetch_ssv_pubkeys(
     let mut pubkeys: Vec<BlsPublicKey> = vec![];
     let mut page = 1;
 
+    // Validate the URL - this appends a trailing slash if missing as efficiently as
+    // possible
+    if !api_url.path().ends_with('/') {
+        match api_url.path_segments_mut() {
+            Ok(mut segments) => segments.push(""), // Analogous to a trailing slash
+            Err(_) => bail!("SSV API URL is not a valid base URL"),
+        };
+    }
+
     loop {
         let route = format!(
             "{chain_name}/validators/in_operator/{node_operator_id}?perPage={MAX_PER_PAGE}&page={page}",
diff --git a/crates/common/src/config/pbs.rs b/crates/common/src/config/pbs.rs
@@ -404,5 +404,5 @@ pub async fn load_pbs_custom_config<T: DeserializeOwned>() -> Result<(PbsModuleC
 
 /// Default URL for the SSV network API
 fn default_ssv_api_url() -> Url {
-    Url::parse("https://api.ssv.network/api/v4").expect("default URL is valid")
+    Url::parse("https://api.ssv.network/api/v4/").expect("default URL is valid")
 }
diff --git a/crates/common/src/pbs/error.rs b/crates/common/src/pbs/error.rs
@@ -25,6 +25,9 @@ pub enum PbsError {
 
     #[error("URL parsing error: {0}")]
     UrlParsing(#[from] url::ParseError),
+
+    #[error("tokio join error: {0}")]
+    TokioJoinError(#[from] tokio::task::JoinError),
 }
 
 impl PbsError {
diff --git a/crates/pbs/src/api.rs b/crates/pbs/src/api.rs
@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use async_trait::async_trait;
 use axum::{Router, http::HeaderMap};
 use cb_common::pbs::{
@@ -34,10 +36,10 @@ pub trait BuilderApi<S: BuilderApiState>: 'static {
     /// https://ethereum.github.io/builder-specs/#/Builder/submitBlindedBlock and
     /// https://ethereum.github.io/builder-specs/#/Builder/submitBlindedBlockV2
     async fn submit_block(
-        signed_blinded_block: SignedBlindedBeaconBlock,
+        signed_blinded_block: Arc<SignedBlindedBeaconBlock>,
         req_headers: HeaderMap,
         state: PbsState<S>,
-        api_version: &BuilderApiVersion,
+        api_version: BuilderApiVersion,
     ) -> eyre::Result<Option<SubmitBlindedBlockResponse>> {
         mev_boost::submit_block(signed_blinded_block, req_headers, state, api_version).await
     }
diff --git a/crates/pbs/src/mev_boost/register_validator.rs b/crates/pbs/src/mev_boost/register_validator.rs
@@ -7,7 +7,10 @@ use cb_common::{
     utils::{get_user_agent_with_version, read_chunked_body_with_max, utcnow_ms},
 };
 use eyre::bail;
-use futures::future::{join_all, select_ok};
+use futures::{
+    FutureExt,
+    future::{join_all, select_ok},
+};
 use reqwest::header::{CONTENT_TYPE, USER_AGENT};
 use tracing::{Instrument, debug, error};
 use url::Url;
@@ -49,32 +52,38 @@ pub async fn register_validator<S: BuilderApiState>(
 
     for (n_regs, body) in bodies {
         for relay in state.all_relays().iter().cloned() {
-            handles.push(tokio::spawn(
-                send_register_validator_with_timeout(
-                    n_regs,
-                    body.clone(),
-                    relay,
-                    send_headers.clone(),
-                    state.pbs_config().timeout_register_validator_ms,
-                    state.pbs_config().register_validator_retry_limit,
+            handles.push(
+                tokio::spawn(
+                    send_register_validator_with_timeout(
+                        n_regs,
+                        body.clone(),
+                        relay,
+                        send_headers.clone(),
+                        state.pbs_config().timeout_register_validator_ms,
+                        state.pbs_config().register_validator_retry_limit,
+                    )
+                    .in_current_span(),
                 )
-                .in_current_span(),
-            ));
+                .map(|join_result| match join_result {
+                    Ok(res) => res,
+                    Err(err) => Err(PbsError::TokioJoinError(err)),
+                }),
+            );
         }
     }
 
     if state.pbs_config().wait_all_registrations {
         // wait for all relays registrations to complete
         let results = join_all(handles).await;
-        if results.into_iter().any(|res| res.is_ok_and(|res| res.is_ok())) {
+        if results.into_iter().any(|res| res.is_ok()) {
             Ok(())
         } else {
             bail!("No relay passed register_validator successfully")
         }
     } else {
         // return once first completes, others proceed in background
-        let result = select_ok(handles).await?;
-        match result.0 {
+        let result = select_ok(handles).await;
+        match result {
             Ok(_) => Ok(()),
             Err(_) => bail!("No relay passed register_validator successfully"),
         }
diff --git a/crates/pbs/src/mev_boost/submit_block.rs b/crates/pbs/src/mev_boost/submit_block.rs
@@ -1,5 +1,6 @@
 use std::{
     str::FromStr,
+    sync::Arc,
     time::{Duration, Instant},
 };
 
@@ -14,7 +15,7 @@ use cb_common::{
     },
     utils::{get_user_agent_with_version, read_chunked_body_with_max, utcnow_ms},
 };
-use futures::future::select_ok;
+use futures::{FutureExt, future::select_ok};
 use reqwest::header::USER_AGENT;
 use tracing::{debug, warn};
 use url::Url;
@@ -31,10 +32,10 @@ use crate::{
 /// https://ethereum.github.io/builder-specs/#/Builder/submitBlindedBlockV2. Use `api_version` to
 /// distinguish between the two.
 pub async fn submit_block<S: BuilderApiState>(
-    signed_blinded_block: SignedBlindedBeaconBlock,
+    signed_blinded_block: Arc<SignedBlindedBeaconBlock>,
     req_headers: HeaderMap,
     state: PbsState<S>,
-    api_version: &BuilderApiVersion,
+    api_version: BuilderApiVersion,
 ) -> eyre::Result<Option<SubmitBlindedBlockResponse>> {
     debug!(?req_headers, "received headers");
 
@@ -58,17 +59,22 @@ pub async fn submit_block<S: BuilderApiState>(
     send_headers.insert(USER_AGENT, get_user_agent_with_version(&req_headers)?);
     send_headers.insert(HEADER_CONSENSUS_VERSION, consensus_version);
 
-    let relays = state.all_relays();
-    let mut handles = Vec::with_capacity(relays.len());
-    for relay in relays.iter() {
-        handles.push(Box::pin(submit_block_with_timeout(
-            &signed_blinded_block,
-            relay,
-            send_headers.clone(),
-            state.pbs_config().timeout_get_payload_ms,
-            api_version,
-            fork_name,
-        )));
+    let mut handles = Vec::with_capacity(state.all_relays().len());
+    for relay in state.all_relays().iter().cloned() {
+        handles.push(
+            tokio::spawn(submit_block_with_timeout(
+                signed_blinded_block.clone(),
+                relay,
+                send_headers.clone(),
+                state.pbs_config().timeout_get_payload_ms,
+                api_version,
+                fork_name,
+            ))
+            .map(|join_result| match join_result {
+                Ok(res) => res,
+                Err(err) => Err(PbsError::TokioJoinError(err)),
+            }),
+        );
     }
 
     let results = select_ok(handles).await;
@@ -81,14 +87,14 @@ pub async fn submit_block<S: BuilderApiState>(
 /// Submit blinded block to relay, retry connection errors until the
 /// given timeout has passed
 async fn submit_block_with_timeout(
-    signed_blinded_block: &SignedBlindedBeaconBlock,
-    relay: &RelayClient,
+    signed_blinded_block: Arc<SignedBlindedBeaconBlock>,
+    relay: RelayClient,
     headers: HeaderMap,
     timeout_ms: u64,
-    api_version: &BuilderApiVersion,
+    api_version: BuilderApiVersion,
     fork_name: ForkName,
 ) -> Result<Option<SubmitBlindedBlockResponse>, PbsError> {
-    let mut url = relay.submit_block_url(*api_version)?;
+    let mut url = relay.submit_block_url(api_version)?;
     let mut remaining_timeout_ms = timeout_ms;
     let mut retry = 0;
     let mut backoff = Duration::from_millis(250);
@@ -97,12 +103,12 @@ async fn submit_block_with_timeout(
         let start_request = Instant::now();
         match send_submit_block(
             url.clone(),
-            signed_blinded_block,
-            relay,
+            &signed_blinded_block,
+            &relay,
             headers.clone(),
             remaining_timeout_ms,
             retry,
-            api_version,
+            &api_version,
             fork_name,
         )
         .await
diff --git a/crates/pbs/src/routes/submit_block.rs b/crates/pbs/src/routes/submit_block.rs
@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use axum::{
     Json,
     extract::State,
@@ -45,11 +47,11 @@ async fn handle_submit_block_impl<S: BuilderApiState, A: BuilderApi<S>>(
     raw_request: RawRequest,
     api_version: BuilderApiVersion,
 ) -> Result<impl IntoResponse, PbsClientError> {
-    let signed_blinded_block =
+    let signed_blinded_block = Arc::new(
         deserialize_body(&req_headers, raw_request.body_bytes).await.map_err(|e| {
             error!(%e, "failed to deserialize signed blinded block");
             PbsClientError::DecodeError(format!("failed to deserialize body: {e}"))
-        })?;
+        })?);
     tracing::Span::current().record("slot", signed_blinded_block.slot().as_u64() as i64);
     tracing::Span::current()
         .record("block_hash", tracing::field::debug(signed_blinded_block.block_hash()));
@@ -75,7 +77,7 @@ async fn handle_submit_block_impl<S: BuilderApiState, A: BuilderApi<S>>(
 
     info!(ua, ms_into_slot = now.saturating_sub(slot_start_ms), "new request");
 
-    match A::submit_block(signed_blinded_block, req_headers, state, &api_version).await {
+    match A::submit_block(signed_blinded_block, req_headers, state, api_version).await {
         Ok(res) => match res {
             Some(payload_and_blobs) => {
                 trace!(?payload_and_blobs);
diff --git a/tests/src/mock_ssv.rs b/tests/src/mock_ssv.rs
@@ -37,7 +37,10 @@ pub async fn create_mock_ssv_server(
         force_timeout: Arc::new(RwLock::new(false)),
     });
     let router = axum::Router::new()
-        .route("/{chain_name}/validators/in_operator/{node_operator_id}", get(handle_validators))
+        .route(
+            "/api/v4/{chain_name}/validators/in_operator/{node_operator_id}",
+            get(handle_validators),
+        )
         .route("/big_data", get(handle_big_data))
         .with_state(state)
         .into_make_service();
diff --git a/tests/tests/pbs_mux.rs b/tests/tests/pbs_mux.rs
@@ -29,8 +29,9 @@ async fn test_ssv_network_fetch() -> Result<()> {
     // Start the mock server
     let port = 30100;
     let _server_handle = create_mock_ssv_server(port, None).await?;
-    let url = Url::parse(&format!("http://localhost:{port}/test_chain/validators/in_operator/1"))
-        .unwrap();
+    let url =
+        Url::parse(&format!("http://localhost:{port}/api/v4/test_chain/validators/in_operator/1"))
+            .unwrap();
     let response =
         fetch_ssv_pubkeys_from_url(url, Duration::from_secs(HTTP_TIMEOUT_SECONDS_DEFAULT)).await?;
 
@@ -100,8 +101,9 @@ async fn test_ssv_network_fetch_timeout() -> Result<()> {
         force_timeout: Arc::new(RwLock::new(true)),
     };
     let server_handle = create_mock_ssv_server(port, Some(state)).await?;
-    let url = Url::parse(&format!("http://localhost:{port}/test_chain/validators/in_operator/1"))
-        .unwrap();
+    let url =
+        Url::parse(&format!("http://localhost:{port}/api/v4/test_chain/validators/in_operator/1"))
+            .unwrap();
     let response = fetch_ssv_pubkeys_from_url(url, Duration::from_secs(TEST_HTTP_TIMEOUT)).await;
 
     // The response should fail due to timeout
diff --git a/tests/tests/pbs_mux_refresh.rs b/tests/tests/pbs_mux_refresh.rs
@@ -43,7 +43,8 @@ async fn test_auto_refresh() -> Result<()> {
 
     // Start the mock SSV API server
     let ssv_api_port = pbs_port + 1;
-    let ssv_api_url = Url::parse(&format!("http://localhost:{ssv_api_port}"))?;
+    // Intentionally missing a trailing slash to ensure this is handled properly
+    let ssv_api_url = Url::parse(&format!("http://localhost:{ssv_api_port}/api/v4"))?;
     let mock_ssv_state = SsvMockState {
         validators: Arc::new(RwLock::new(vec![SSVValidator {
             pubkey: existing_mux_pubkey.clone(),

Original file line number	Diff line number	Diff line change
`@@ -404,5 +404,5 @@ pub async fn load_pbs_custom_config<T: DeserializeOwned>() -> Result<(PbsModuleC`
`404`	`404`
`405`	`405`	`/// Default URL for the SSV network API`
`406`	`406`	`fn default_ssv_api_url() -> Url {`
`407`		`- Url::parse("https://api.ssv.network/api/v4").expect("default URL is valid")`
	`407`	`+ Url::parse("https://api.ssv.network/api/v4/").expect("default URL is valid")`
`408`	`408`	`}`
Original file line number	Diff line number	Diff line change
`@@ -25,6 +25,9 @@ pub enum PbsError {`
`25`	`25`
`26`	`26`	`#[error("URL parsing error: {0}")]`
`27`	`27`	`UrlParsing(#[from] url::ParseError),`
	`28`	`+`
	`29`	`+ #[error("tokio join error: {0}")]`
	`30`	`+ TokioJoinError(#[from] tokio::task::JoinError),`
`28`	`31`	`}`
`29`	`32`
`30`	`33`	`impl PbsError {`