http fallback

Author: ninaiiad

crates/common/src/pbs/error.rs

@@ -42,6 +42,9 @@ pub enum PbsError {
     #[error("websocket error: {0}")]
     WebSocket(String),
 
+    #[error("websocket connect failed: {0}")]
+    WebSocketConnect(String),
+
     #[error("websocket timed out")]
     WebSocketTimeout,
 }

crates/pbs/src/mev_boost/get_header.rs

@@ -251,7 +251,29 @@ async fn get_header_from_relay(
 
     match relay.get_header_request(params.slot, &params.parent_hash, &params.pubkey)? {
         GetHeaderRequest::Stream(url) => {
-            get_header_ws(&request_info, &relay, url, timeout_left_ms).await
+            let started = Instant::now();
+            let err = match get_header_ws(&request_info, &relay, url, timeout_left_ms).await {
+                Err(PbsError::WebSocketConnect(err)) => err,
+                res => return res,
+            };
+
+            let elapsed_ms = started.elapsed().as_millis() as u64;
+            let timeout_left_ms = timeout_left_ms.saturating_sub(elapsed_ms);
+            if timeout_left_ms == 0 {
+                return Err(PbsError::WebSocketConnect(err));
+            }
+
+            warn!(relay_id = relay.id.as_ref(), %err, timeout_left_ms, "stream failed, falling back to http get_header");
+
+            let url = relay.get_header_url(params.slot, &params.parent_hash, &params.pubkey)?;
+            send_timed_get_header(
+                request_info,
+                relay,
+                ms_into_slot + elapsed_ms,
+                url,
+                timeout_left_ms,
+            )
+            .await
         }
         GetHeaderRequest::Http(url) => {
             send_timed_get_header(request_info, relay, ms_into_slot, url, timeout_left_ms).await
@@ -271,7 +293,7 @@ async fn send_timed_get_header(
             // sleep until target time in slot
 
             let delay = target_ms.saturating_sub(ms_into_slot);
-            if delay > 0 {
+            if delay > 0 && delay < timeout_left_ms {
                 debug!(
                     relay_id = relay.id.as_ref(),
                     target_ms, ms_into_slot, "TG: waiting to send first header request"
@@ -281,7 +303,10 @@ async fn send_timed_get_header(
             } else {
                 debug!(
                     relay_id = relay.id.as_ref(),
-                    target_ms, ms_into_slot, "TG: request already late enough in slot"
+                    target_ms,
+                    ms_into_slot,
+                    timeout_left_ms,
+                    "TG: sending first header request now"
                 );
             }
         }

crates/pbs/src/mev_boost/get_header_ws.rs

@@ -101,7 +101,7 @@ pub(super) async fn get_header_ws(
             let status = rejected.as_ref().map_or(TRANSPORT_ERROR_CODE_STR, |code| code.as_str());
 
             record_status(status, relay);
-            return Err(PbsError::WebSocket(format!("connect failed: {err}")));
+            return Err(PbsError::WebSocketConnect(err.to_string()));
         }
         Err(_) => {
             record_status(TIMEOUT_ERROR_CODE_STR, relay);
@@ -215,7 +215,7 @@ fn build_handshake_request(
     let mut request = url
         .as_str()
         .into_client_request()
-        .map_err(|err| PbsError::WebSocket(format!("invalid ws url: {err}")))?;
+        .map_err(|err| PbsError::WebSocketConnect(format!("invalid ws url: {err}")))?;
 
     let headers = request.headers_mut();
     if let Some(user_agent) = request_info.headers.get(USER_AGENT) {

tests/src/mock_relay.rs

@@ -1,5 +1,5 @@
 use std::{
-    collections::HashSet,
+    collections::{HashMap, HashSet},
     net::SocketAddr,
     sync::{
         Arc, RwLock,
@@ -36,7 +36,10 @@ use cb_common::{
         get_consensus_version_header, get_content_type,
     },
 };
-use cb_pbs::MAX_SIZE_SUBMIT_BLOCK_RESPONSE;
+use cb_pbs::{
+    GET_HEADER_ENDPOINT_TAG, MAX_SIZE_SUBMIT_BLOCK_RESPONSE, REGISTER_VALIDATOR_ENDPOINT_TAG,
+    STATUS_ENDPOINT_TAG, SUBMIT_BLINDED_BLOCK_ENDPOINT_TAG,
+};
 use lh_types::KzgProof;
 use reqwest::header::{ACCEPT, CONTENT_TYPE};
 use ssz::Encode;
@@ -95,7 +98,9 @@ pub struct MockRelayState {
     /// The raw `Accept` header PBS sent on the most recent get_header request,
     /// so a test can assert what encoding PBS asked the relay for.
     received_get_header_accept: RwLock<Option<String>>,
-    last_register_api_key: RwLock<Option<String>>,
+    /// Api key header seen per endpoint tag, so a test can assert the relay's
+    /// configured key rides on every request PBS sends it.
+    api_keys_seen: RwLock<HashMap<&'static str, String>>,
 }
 
 impl MockRelayState {
@@ -135,8 +140,14 @@ impl MockRelayState {
     pub fn set_response_override(&self, status: StatusCode) {
         *self.response_override.write().unwrap() = Some(status);
     }
-    pub fn last_register_api_key(&self) -> Option<String> {
-        self.last_register_api_key.read().unwrap().clone()
+    /// Api key seen on `endpoint`, one of the `*_ENDPOINT_TAG` constants
+    pub fn api_key_seen(&self, endpoint: &str) -> Option<String> {
+        self.api_keys_seen.read().unwrap().get(endpoint).cloned()
+    }
+    fn record_api_key(&self, endpoint: &'static str, headers: &HeaderMap) {
+        if let Some(api_key) = headers.get(HEADER_API_KEY).and_then(|key| key.to_str().ok()) {
+            self.api_keys_seen.write().unwrap().insert(endpoint, api_key.to_string());
+        }
     }
 }
 
@@ -158,7 +169,7 @@ impl MockRelayState {
             response_override: RwLock::new(None),
             bid_value: RwLock::new(U256::from(10)),
             received_get_header_accept: RwLock::new(None),
-            last_register_api_key: RwLock::new(None),
+            api_keys_seen: RwLock::new(HashMap::new()),
             supported_content_types: Arc::new(
                 [EncodingType::Json, EncodingType::Ssz].iter().cloned().collect(),
             ),
@@ -260,6 +271,7 @@ async fn handle_get_header(
     headers: HeaderMap,
 ) -> Response {
     state.received_get_header.fetch_add(1, Ordering::Relaxed);
+    state.record_api_key(GET_HEADER_ENDPOINT_TAG, &headers);
     *state.received_get_header_accept.write().unwrap() =
         headers.get(ACCEPT).and_then(|v| v.to_str().ok()).map(String::from);
     let accept_types = get_accept_types(&headers)
@@ -322,8 +334,12 @@ async fn handle_get_header(
     response
 }
 
-async fn handle_get_status(State(state): State<Arc<MockRelayState>>) -> impl IntoResponse {
+async fn handle_get_status(
+    State(state): State<Arc<MockRelayState>>,
+    headers: HeaderMap,
+) -> impl IntoResponse {
     state.received_get_status.fetch_add(1, Ordering::Relaxed);
+    state.record_api_key(STATUS_ENDPOINT_TAG, &headers);
     // Production `get_status` dispatches relays concurrently via `select_ok`,
     // which cancels losing futures as soon as any relay returns OK. On a
     // loaded runner this can abort a sibling relay's reqwest send before
@@ -341,10 +357,7 @@ async fn handle_register_validator(
     Json(validators): Json<Vec<ValidatorRegistration>>,
 ) -> impl IntoResponse {
     state.received_register_validator.fetch_add(1, Ordering::Relaxed);
-    *state.last_register_api_key.write().unwrap() = headers
-        .get(HEADER_API_KEY)
-        .and_then(|value| value.to_str().ok())
-        .map(|value| value.to_string());
+    state.record_api_key(REGISTER_VALIDATOR_ENDPOINT_TAG, &headers);
     debug!("Received {} registrations", validators.len());
 
     if let Some(status) = state.response_override.read().unwrap().as_ref() {
@@ -363,6 +376,7 @@ async fn handle_submit_block_v1(
         return StatusCode::NOT_FOUND.into_response();
     }
     state.received_submit_block.fetch_add(1, Ordering::Relaxed);
+    state.record_api_key(SUBMIT_BLINDED_BLOCK_ENDPOINT_TAG, &headers);
     // Short-circuit SSZ requests with an overridden status so tests can
     // drive the PBS SSZ→JSON retry logic. JSON requests still take the
     // normal path so a single mock run can exercise both attempts.
@@ -475,6 +489,7 @@ async fn handle_submit_block_v2(
         return StatusCode::NOT_FOUND.into_response();
     }
     state.received_submit_block.fetch_add(1, Ordering::Relaxed);
+    state.record_api_key(SUBMIT_BLINDED_BLOCK_ENDPOINT_TAG, &headers);
     // See comment in `handle_submit_block_v1`. Override SSZ with the
     // injected status so C3 tests can assert retry / no-retry behavior.
     if let Some(status) = state.submit_block_ssz_status_override() &&

tests/src/utils.rs

@@ -25,6 +25,9 @@ use url::Url;
 
 pub const HEADER_API_KEY: &str = "x-api-key";
 pub const API_KEY: &str = "123e4567-e89b-12d3-a456-426614174000";
+/// Distinct from [`API_KEY`], which `MockValidator` also sends to PBS: a relay
+/// that sees this one can only have got it from its own config.
+pub const RELAY_API_KEY: &str = "f81d4fae-7dec-11d0-a765-00a0c91e6bf6";
 
 pub fn get_local_address(port: u16) -> String {
     format!("http://0.0.0.0:{port}")
@@ -74,12 +77,36 @@ pub fn generate_mock_relay_with_batch_size(
     RelayClient::new(config)
 }
 
+pub fn generate_mock_relay_with_api_key(
+    port: u16,
+    pubkey: BlsPublicKey,
+    api_key: &str,
+) -> Result<RelayClient> {
+    let mut config = mock_relay_config(port, pubkey)?;
+    config.headers = Some(HashMap::from([(HEADER_API_KEY.into(), api_key.into())]));
+    RelayClient::new(config)
+}
+
 pub fn generate_mock_stream_relay(port: u16, pubkey: BlsPublicKey) -> Result<RelayClient> {
     let mut config = mock_relay_config(port, pubkey)?;
     config.get_header = GetHeaderTransport::Stream;
     RelayClient::new(config)
 }
 
+pub fn generate_mock_stream_relay_with_timing_games(
+    port: u16,
+    pubkey: BlsPublicKey,
+    target_first_request_ms: u64,
+    frequency_get_header_ms: u64,
+) -> Result<RelayClient> {
+    let mut config = mock_relay_config(port, pubkey)?;
+    config.get_header = GetHeaderTransport::Stream;
+    config.enable_timing_games = true;
+    config.target_first_request_ms = Some(target_first_request_ms);
+    config.frequency_get_header_ms = Some(frequency_get_header_ms);
+    RelayClient::new(config)
+}
+
 pub fn get_pbs_config(port: u16) -> PbsConfig {
     PbsConfig {
         host: Ipv4Addr::UNSPECIFIED,

tests/tests/pbs_get_header_ws.rs

@@ -19,8 +19,9 @@ use cb_tests::{
     mock_validator::MockValidator,
     mock_ws_relay::{MockWsRelayState, start_mock_ws_relay_service},
     utils::{
-        API_KEY, generate_mock_relay, generate_mock_stream_relay, get_free_listener,
-        get_pbs_config, setup_test_env, to_pbs_config,
+        API_KEY, generate_mock_relay, generate_mock_stream_relay,
+        generate_mock_stream_relay_with_timing_games, get_free_listener, get_pbs_config,
+        setup_test_env, to_pbs_config,
     },
 };
 use eyre::Result;
@@ -290,6 +291,62 @@ async fn test_get_header_ws_unreachable_relay_returns_204() -> Result<()> {
     Ok(())
 }
 
+/// A relay configured to stream that rejects the handshake — here by not
+/// serving the stream route at all — is still asked over HTTP, so its bid stays
+/// in the auction instead of the relay silently dropping out.
+#[tokio::test]
+async fn test_get_header_ws_falls_back_to_http() -> Result<()> {
+    setup_test_env();
+    let signer = random_secret();
+    let pubkey = signer.public_key();
+    let chain = Chain::Hoodi;
+
+    let listener = get_free_listener().await;
+    let port = listener.local_addr()?.port();
+    let relay_state =
+        Arc::new(MockRelayState::new(chain, signer.clone()).with_bid_value(U256::from(42)));
+    tokio::spawn(start_mock_relay_service_with_listener(relay_state.clone(), listener));
+
+    let relay = generate_mock_stream_relay(port, pubkey)?;
+    let validator = start_pbs(chain, vec![relay], 1_000).await?;
+
+    let (code, res) = get_header_json(&validator).await?;
+    assert_eq!(code, StatusCode::OK);
+    assert_bid(&res.unwrap(), chain, &signer, U256::from(42));
+
+    // Timing games are off for this relay, so the fallback is a single request
+    assert_eq!(relay_state.received_get_header(), 1);
+    Ok(())
+}
+
+/// The handshake can fail early in the slot, so the fallback is a normal http
+/// get_header and still plays this relay's timing games.
+#[tokio::test]
+async fn test_get_header_ws_fallback_runs_timing_games() -> Result<()> {
+    setup_test_env();
+    let signer = random_secret();
+    let pubkey = signer.public_key();
+    let chain = Chain::Hoodi;
+
+    let listener = get_free_listener().await;
+    let port = listener.local_addr()?.port();
+    let relay_state = Arc::new(MockRelayState::new(chain, signer.clone()));
+    tokio::spawn(start_mock_relay_service_with_listener(relay_state.clone(), listener));
+
+    // Target is already behind us, so no wait, then one request per 200ms of
+    // what is left of the budget
+    let relay = generate_mock_stream_relay_with_timing_games(port, pubkey, 0, 200)?;
+    let validator = start_pbs(chain, vec![relay], 1_000).await?;
+
+    let (code, res) = get_header_json(&validator).await?;
+    assert_eq!(code, StatusCode::OK);
+    assert_bid(&res.unwrap(), chain, &signer, U256::from(10));
+
+    let n_requests = relay_state.received_get_header();
+    assert!(n_requests > 1, "fallback skipped the timing games loop: {n_requests} requests");
+    Ok(())
+}
+
 /// A streamed bid competes in the same auction as an HTTP one
 #[tokio::test]
 async fn test_get_header_ws_wins_auction_against_http() -> Result<()> {

tests/tests/pbs_post_validators.rs

@@ -7,14 +7,9 @@ use cb_common::{
 };
 use cb_pbs::{DefaultBuilderApi, PbsService, PbsState};
 use cb_tests::{
-    mock_relay::{
-        MockRelayState, start_mock_relay_service, start_mock_relay_service_with_listener,
-    },
+    mock_relay::{MockRelayState, start_mock_relay_service},
     mock_validator::MockValidator,
-    utils::{
-        API_KEY, generate_mock_relay, get_free_listener, get_pbs_config, setup_test_env,
-        to_pbs_config,
-    },
+    utils::{generate_mock_relay, get_pbs_config, setup_test_env, to_pbs_config},
 };
 use eyre::Result;
 use reqwest::StatusCode;
@@ -66,38 +61,6 @@ async fn test_register_validators() -> Result<()> {
     Ok(())
 }
 
-/// The api key from `relay.config.headers` is forwarded on registrations.
-#[tokio::test]
-async fn test_register_validators_sends_api_key() -> Result<()> {
-    setup_test_env();
-    let signer = random_secret();
-    let pubkey: BlsPublicKey = signer.public_key();
-    let chain = Chain::Holesky;
-
-    let relay_listener = get_free_listener().await;
-    let relay_port = relay_listener.local_addr()?.port();
-    let mock_state = Arc::new(MockRelayState::new(chain, signer));
-    tokio::spawn(start_mock_relay_service_with_listener(mock_state.clone(), relay_listener));
-
-    let pbs_listener = get_free_listener().await;
-    let pbs_port = pbs_listener.local_addr()?.port();
-    drop(pbs_listener);
-
-    let relays = vec![generate_mock_relay(relay_port, pubkey)?];
-    let config = to_pbs_config(chain, get_pbs_config(pbs_port), relays);
-    tokio::spawn(PbsService::run::<(), DefaultBuilderApi>(PbsState::new(config, PathBuf::new())));
-    tokio::time::sleep(Duration::from_millis(100)).await;
-
-    let mock_validator = MockValidator::new(pbs_port)?;
-    let res = mock_validator.do_register_validator().await?;
-    assert_eq!(res.status(), StatusCode::OK);
-
-    assert_eq!(mock_state.received_register_validator(), 1);
-    assert_eq!(mock_state.last_register_api_key().as_deref(), Some(API_KEY));
-
-    Ok(())
-}
-
 #[tokio::test]
 async fn test_register_validators_does_not_retry_on_429() -> Result<()> {
     setup_test_env();