ws get_header bids stream draft (#483)

Author: ninaiiad

Cargo.lock

@@ -79,6 +79,7 @@ subtle = "2.5"
 tempfile = "3.20.0"
 thiserror = "2.0.12"
 tokio = { version = "1.37.0", features = ["full"] }
+tokio-tungstenite = { version = "0.28.0", default-features = false, features = ["connect", "rustls-tls-webpki-roots"] }
 toml = "0.8.13"
 tonic = { version = "0.12.3", features = ["channel", "prost", "tls"] }
 tonic-build = "0.12.3"
@@ -93,6 +94,7 @@ typenum = "1.17.0"
 unicode-normalization = "0.1.24"
 url = { version = "2.5.0", features = ["serde"] }
 uuid = { version = "1.8.0", features = ["fast-rng", "serde", "v4"] }
+webpki-roots = "1.0"
 
 [patch.crates-io]
 blstrs_plus = { git = "https://github.com/Commit-Boost/blstrs" }

Cargo.toml

@@ -44,7 +44,7 @@ use cb_common::{pbs::GetHeaderParams, signer::random_secret, types::Chain};
 use cb_pbs::{PbsState, get_header};
 use cb_tests::{
     mock_relay::{MockRelayState, start_mock_relay_service},
-    utils::{generate_mock_relay, get_pbs_static_config, to_pbs_config},
+    utils::{generate_mock_relay, get_pbs_config, to_pbs_config},
 };
 use criterion::{Criterion, black_box, criterion_group, criterion_main};
 
@@ -103,8 +103,7 @@ fn bench_get_header(c: &mut Criterion) {
         let states: Vec<PbsState> = RELAY_COUNTS
             .iter()
             .map(|&n| {
-                let config =
-                    to_pbs_config(CHAIN, get_pbs_static_config(0), relay_clients[..n].to_vec());
+                let config = to_pbs_config(CHAIN, get_pbs_config(0), relay_clients[..n].to_vec());
                 PbsState::new(config, PathBuf::new())
             })
             .collect();

benches/microbench/src/get_header.rs

@@ -2,7 +2,7 @@ use std::time::{Duration, Instant};
 
 use alloy::primitives::B256;
 use cb_common::{
-    config::RelayConfig,
+    config::{GetHeaderTransport, RelayConfig},
     pbs::{GetHeaderResponse, RelayClient, RelayEntry},
     types::{BlsPublicKey, BlsSecretKey, Chain},
     utils::TestRandomSeed,
@@ -157,6 +157,7 @@ fn get_mock_validator(bench: BenchConfig) -> RelayClient {
         id: None,
         headers: None,
         get_params: None,
+        get_header: GetHeaderTransport::Http,
         enable_timing_games: false,
         target_first_request_ms: None,
         frequency_get_header_ms: None,

benches/pbs/src/main.rs

@@ -92,6 +92,15 @@ headers = { X-MyCustomHeader = "MyCustomValue" }
 # GET parameters to add to each request URL for this relay
 # OPTIONAL
 get_params = { param1 = "value1", param2 = "value2" }
+# How to fetch headers from this relay.
+#   "http"     -> one request per get_header, at the relay url above
+#   "stream"   -> websocket stream of bid updates, only for relays that support it. Connects to
+#                 ws(s)://<relay url>/eth/v1/builder/header_stream/{slot}/{parent_hash}/{pubkey}.
+#                 Requires a UUID api key, set it in `headers` field above under the `X-Api-Key`
+#                 name; it is sent on the websocket handshake. Can use 
+#                 https://www.uuidgenerator.net/version4 to generate one.
+# OPTIONAL, DEFAULT: "http"
+get_header = "http"
 # Whether to enable timing games, as tuned by `target_first_request_ms` and `frequency_get_header_ms`.
 # NOTE: if neither `target_first_request_ms` nor `frequency_get_header_ms` is set, this flag has no effect.
 #

config.example.toml

@@ -55,6 +55,7 @@ tree_hash.workspace = true
 tree_hash_derive.workspace = true
 unicode-normalization.workspace = true
 url.workspace = true
+uuid.workspace = true
 reqwest-eventsource.workspace = true
 
 [dev-dependencies]

crates/common/Cargo.toml

@@ -38,6 +38,14 @@ use crate::{
     },
 };
 
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Deserialize, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum GetHeaderTransport {
+    #[default]
+    Http,
+    Stream,
+}
+
 #[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(deny_unknown_fields)]
 pub struct RelayConfig {
@@ -50,6 +58,9 @@ pub struct RelayConfig {
     pub headers: Option<HashMap<String, String>>,
     /// Optional GET parameters to add to each request
     pub get_params: Option<HashMap<String, String>>,
+    /// How to fetch headers from this relay
+    #[serde(default)]
+    pub get_header: GetHeaderTransport,
     /// Whether to enable timing games
     #[serde(default = "default_bool::<false>")]
     pub enable_timing_games: bool,

crates/common/src/config/pbs.rs

@@ -4,6 +4,8 @@ pub const BUILDER_V1_API_PATH: &str = "/eth/v1/builder";
 pub const BUILDER_V2_API_PATH: &str = "/eth/v2/builder";
 
 pub const GET_HEADER_PATH: &str = "/header/{slot}/{parent_hash}/{pubkey}";
+
+pub const GET_HEADER_STREAM_PATH: &str = "/header_stream";
 pub const GET_STATUS_PATH: &str = "/status";
 pub const REGISTER_VALIDATOR_PATH: &str = "/validators";
 pub const SUBMIT_BLOCK_PATH: &str = "/blinded_blocks";
@@ -17,6 +19,7 @@ pub const HEADER_VERSION_KEY: &str = "X-CommitBoost-Version";
 pub const HEADER_VERSION_VALUE: &str = COMMIT_BOOST_VERSION;
 pub const HEADER_START_TIME_UNIX_MS: &str = "Date-Milliseconds";
 pub const HEADER_TIMEOUT_MS: &str = "X-Timeout-Ms";
+pub const HEADER_API_KEY: &str = "X-Api-Key";
 pub const HEADER_CONSENSUS_VERSION: &str = "Eth-Consensus-Version";
 
 pub const DEFAULT_PBS_JWT_KEY: &str = "DEFAULT_PBS";

crates/common/src/pbs/constants.rs

@@ -38,11 +38,24 @@ pub enum PbsError {
 
     #[error("SSZ error: {0}")]
     SszError(#[from] SszValueError),
+
+    #[error("websocket error: {0}")]
+    WebSocket(String),
+
+    #[error("websocket connect failed: {0}")]
+    WebSocketConnect(String),
+
+    #[error("websocket timed out")]
+    WebSocketTimeout,
 }
 
 impl PbsError {
     pub fn is_timeout(&self) -> bool {
-        matches!(self, PbsError::Reqwest(err) if err.is_timeout())
+        match self {
+            PbsError::Reqwest(err) => err.is_timeout(),
+            PbsError::WebSocketTimeout => true,
+            _ => false,
+        }
     }
 
     /// Extract the HTTP status code from relay-originated errors.