diff --git a/Cargo.lock b/Cargo.lock index 140c3303..32e64a92 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1992,6 +1992,7 @@ dependencies = [ "async-trait", "axum 0.8.9", "axum-extra", + "base64 0.22.1", "cb-common", "cb-metrics", "ethereum_serde_utils 0.7.0", @@ -2056,6 +2057,7 @@ version = "0.11.0" dependencies = [ "alloy", "axum 0.8.9", + "base64 0.22.1", "cb-common", "cb-pbs", "cb-signer", diff --git a/config.example.toml b/config.example.toml index 1c6354e6..430050a1 100644 --- a/config.example.toml +++ b/config.example.toml @@ -107,6 +107,7 @@ get_params = { param1 = "value1", param2 = "value2" } # 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. +# (unreleased, from v0.12.0-rc1) After Gloas this also streams ePBS bids (https://commit-boost.github.io/commit-boost-client/get_started/epbs#bid-streaming) # OPTIONAL, DEFAULT: "http" get_header = "http" # Whether to enable timing games, as tuned by `target_first_request_ms` and `frequency_get_header_ms`. diff --git a/crates/common/src/pbs/constants.rs b/crates/common/src/pbs/constants.rs index e7639530..111d77e9 100644 --- a/crates/common/src/pbs/constants.rs +++ b/crates/common/src/pbs/constants.rs @@ -13,6 +13,7 @@ pub const RELOAD_PATH: &str = "/reload"; pub const GET_EXECUTION_PAYLOAD_BID_PATH: &str = "/execution_payload_bid/{slot}/{parent_hash}/{parent_root}/{proposer_pubkey}"; +pub const GET_EXECUTION_PAYLOAD_BID_STREAM_PATH: &str = "/execution_payload_bid_stream"; pub const SUBMIT_BUILDER_PREFERENCES_PATH: &str = "/builder_preferences/{proposer_pubkey}"; pub const SUBMIT_SIGNED_BEACON_BLOCK_PATH: &str = "/beacon_blocks"; @@ -24,6 +25,9 @@ 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"; +/// The SSZ `SignedBuilderRequestAuth` as padded standard base64, on the ePBS +/// bid stream handshake: a websocket upgrade has no body +pub const HEADER_REQUEST_AUTH: &str = "X-Request-Auth"; pub const HEADER_API_KEY: &str = "X-Api-Key"; pub const HEADER_CONSENSUS_VERSION: &str = "Eth-Consensus-Version"; diff --git a/crates/common/src/pbs/relay.rs b/crates/common/src/pbs/relay.rs index 4dd48e40..f6878af8 100644 --- a/crates/common/src/pbs/relay.rs +++ b/crates/common/src/pbs/relay.rs @@ -11,8 +11,8 @@ use url::Url; use super::{ HEADER_VERSION_KEY, HEADER_VERSION_VALUE, constants::{ - GET_HEADER_STREAM_PATH, GET_STATUS_PATH, REGISTER_VALIDATOR_PATH, SUBMIT_BLOCK_PATH, - SUBMIT_SIGNED_BEACON_BLOCK_PATH, + GET_EXECUTION_PAYLOAD_BID_STREAM_PATH, GET_HEADER_STREAM_PATH, GET_STATUS_PATH, + REGISTER_VALIDATOR_PATH, SUBMIT_BLOCK_PATH, SUBMIT_SIGNED_BEACON_BLOCK_PATH, }, error::PbsError, }; @@ -89,7 +89,7 @@ pub struct RelayClient { pub client: reqwest::Client, /// Base url of the get_header stream, `Some` only when the relay streams. stream_url: Option, - /// Baseline headers for the get_header stream handshake. + /// Baseline headers for the bid stream handshakes. stream_headers: Arc, /// Configuration of the relay pub config: Arc, @@ -244,6 +244,24 @@ impl RelayClient { ) } + /// The ePBS bid stream, at the get_header stream's origin. `None` when the + /// relay does not stream. + pub fn get_execution_payload_bid_stream_url( + &self, + slot: u64, + parent_hash: &B256, + parent_root: &B256, + validator_pubkey: &BlsPublicKey, + ) -> Option { + let mut url = self.stream_url.clone()?; + url.set_path(&format!( + "{}{GET_EXECUTION_PAYLOAD_BID_STREAM_PATH}/{slot}/{parent_hash}/{parent_root}/{validator_pubkey}", + BuilderApiVersion::V1.path() + )); + self.append_get_params(&mut url); + Some(url) + } + /// builder-API: POST /eth/v1/builder/builder_preferences/{proposer_pubkey} pub fn submit_builder_preferences_url( &self, @@ -460,6 +478,41 @@ mod tests { assert!(RelayClient::new(config).is_err()); } + #[test] + fn test_execution_payload_bid_stream_url() { + let parent_hash = B256::ZERO; + let parent_root = B256::repeat_byte(1); + let validator_pubkey = bls_pubkey_from_hex_unchecked( + "0xac6e77dfe25ecd6110b8e780608cce0dab71fdd5ebea22a16c0205200f2f8e2e3ad3b71d3499c54ad14d6c21b41a37ae", + ); + let relay_config = r#" + { + "url": "https://0xa1cec75a3f0661e99299274182938151e8433c61a19222347ea1313d839229cb4ce4e3e5aa2bdeb71c8fcf1b084963c2@abc.xyz:4444/relay-api" + }"#; + let mut config = serde_json::from_str::(relay_config).unwrap(); + let url = |config: &RelayConfig| { + RelayClient::new(config.clone()).unwrap().get_execution_payload_bid_stream_url( + 0, + &parent_hash, + &parent_root, + &validator_pubkey, + ) + }; + + // Only a streaming relay has one + assert!(url(&config).is_none()); + + // The get_header stream's origin, at the fixed path, with get_params + config.get_header = GetHeaderTransport::Stream; + config.get_params = Some(HashMap::from([("token".to_string(), "abc".to_string())])); + assert_eq!( + url(&config).unwrap().to_string(), + format!( + "wss://abc.xyz:4444/eth/v1/builder/execution_payload_bid_stream/0/{parent_hash}/{parent_root}/{validator_pubkey}?token=abc" + ) + ); + } + #[test] fn test_value_fingerprint_tracks_the_value() { assert_eq!(value_fingerprint("s3cret").len(), 8); diff --git a/crates/pbs/Cargo.toml b/crates/pbs/Cargo.toml index 7037a566..cdc46a19 100644 --- a/crates/pbs/Cargo.toml +++ b/crates/pbs/Cargo.toml @@ -13,6 +13,7 @@ alloy.workspace = true async-trait.workspace = true axum.workspace = true axum-extra.workspace = true +base64.workspace = true cb-common.workspace = true cb-metrics.workspace = true ethereum_serde_utils.workspace = true diff --git a/crates/pbs/src/constants.rs b/crates/pbs/src/constants.rs index 5790af41..a48e07e3 100644 --- a/crates/pbs/src/constants.rs +++ b/crates/pbs/src/constants.rs @@ -9,6 +9,7 @@ pub const SUBMIT_BUILDER_PREFERENCES_ENDPOINT_TAG: &str = "submit_builder_prefer pub const SUBMIT_SIGNED_BEACON_BLOCK_ENDPOINT_TAG: &str = "submit_signed_beacon_block"; pub const RELOAD_ENDPOINT_TAG: &str = "reload"; pub const GET_HEADER_STREAM_ENDPOINT_TAG: &str = "get_header_stream"; +pub const GET_EXECUTION_PAYLOAD_BID_STREAM_ENDPOINT_TAG: &str = "get_execution_payload_bid_stream"; /// For metrics recorded when a request times out pub const TIMEOUT_ERROR_CODE: u16 = 555; diff --git a/crates/pbs/src/dial.rs b/crates/pbs/src/dial.rs index 62be496e..55aac192 100644 --- a/crates/pbs/src/dial.rs +++ b/crates/pbs/src/dial.rs @@ -205,6 +205,7 @@ fn dial_config(url: Url) -> RelayConfig { id: None, headers: None, get_params: None, + // HTTP only: a stream's connection skips DialResolver's address check get_header: GetHeaderTransport::Http, enable_timing_games: false, target_first_request_ms: None, @@ -329,6 +330,7 @@ mod tests { (Ok(relay), Ok(url)) => { assert_eq!(relay.config.entry.url.as_str(), url, "{case}"); assert_eq!(relay.id.as_str(), "dial", "{case}"); + assert_eq!(relay.config.get_header, GetHeaderTransport::Http, "{case}"); } (Err(err), Err(message)) => assert_eq!(err.to_string(), message, "{case}"), _ => panic!("{case}: expected {expected:?}, got {:?}", res.map(|_| ())), diff --git a/crates/pbs/src/metrics.rs b/crates/pbs/src/metrics.rs index 51846f94..a16491aa 100644 --- a/crates/pbs/src/metrics.rs +++ b/crates/pbs/src/metrics.rs @@ -96,6 +96,50 @@ lazy_static! { ) .unwrap(); + // THE ePBS BID STREAM + // Outcome and time-to-first-bid ride RELAY_STATUS_CODE / RELAY_LATENCY + // under `get_execution_payload_bid_stream` + + /// ePBS bid stream handshake latency by relay + pub static ref RELAY_EPBS_STREAM_CONNECT_LATENCY: HistogramVec = register_histogram_vec_with_registry!( + "relay_epbs_stream_connect_latency", + "ePBS bid stream handshake latency by relay", + &["relay_id"], + vec![0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.15, 0.2, 0.3, 0.4, 0.5, 1.0], + PBS_METRICS_REGISTRY + ) + .unwrap(); + + /// Bids that decoded per ePBS bid stream, by relay + pub static ref RELAY_EPBS_STREAM_UPDATES: HistogramVec = register_histogram_vec_with_registry!( + "relay_epbs_stream_updates", + "Bids that decoded per ePBS bid stream, by relay", + &["relay_id"], + vec![0.0, 1.0, 2.0, 3.0, 5.0, 10.0, 20.0, 50.0, 100.0], + PBS_METRICS_REGISTRY + ) + .unwrap(); + + /// ePBS bid stream frames that could not be read or decoded, by relay + pub static ref RELAY_EPBS_STREAM_INVALID_FRAMES: IntCounterVec = register_int_counter_vec_with_registry!( + "relay_epbs_stream_invalid_frames_total", + "ePBS bid stream frames that could not be read or decoded, by relay", + &["relay_id"], + PBS_METRICS_REGISTRY + ) + .unwrap(); + + /// ePBS bid streams that fell back to HTTP, by relay + // Only a failed connect with time left retries; one at the deadline shows + // on the status series alone + pub static ref RELAY_EPBS_STREAM_FALLBACK: IntCounterVec = register_int_counter_vec_with_registry!( + "relay_epbs_stream_fallback_total", + "ePBS bid streams that fell back to HTTP, by relay", + &["relay_id"], + PBS_METRICS_REGISTRY + ) + .unwrap(); + // TO BEACON NODE /// Status code returned to beacon node by endpoint pub static ref BEACON_NODE_STATUS: IntCounterVec = register_int_counter_vec_with_registry!( diff --git a/crates/pbs/src/routes/execution_payload_bid.rs b/crates/pbs/src/routes/execution_payload_bid.rs index f4b57bd7..444ec5b8 100644 --- a/crates/pbs/src/routes/execution_payload_bid.rs +++ b/crates/pbs/src/routes/execution_payload_bid.rs @@ -8,9 +8,9 @@ use axum::{ }; use cb_common::{ pbs::{ - GetExecutionPayloadBidParams, GetExecutionPayloadBidResponse, HEADER_START_TIME_UNIX_MS, - HEADER_TIMEOUT_MS, RelayClient, SignedBuilderRequestAuth, SignedExecutionPayloadBid, - error::PbsError, + ForkName, GetExecutionPayloadBidParams, GetExecutionPayloadBidResponse, + HEADER_START_TIME_UNIX_MS, HEADER_TIMEOUT_MS, RelayClient, SignedBuilderRequestAuth, + SignedExecutionPayloadBid, error::PbsError, }, utils::{ms_into_slot, utcnow_ms}, wire::{ @@ -24,15 +24,18 @@ use reqwest::{ header::{ACCEPT, CONTENT_TYPE}, }; use ssz::{Decode, Encode}; +use tokio::time::Instant; use tracing::{debug, error, info, warn}; +use url::Url; +use super::execution_payload_bid_ws::get_execution_payload_bid_ws; use crate::{ PbsStateGuard, constants::{ GET_EXECUTION_PAYLOAD_BID_ENDPOINT_TAG, MAX_SIZE_DEFAULT, MAX_SIZE_GET_HEADER_RESPONSE, }, error::PbsClientError, - metrics::{RELAY_HEADER_VALUE, RELAY_LAST_SLOT}, + metrics::{RELAY_EPBS_STREAM_FALLBACK, RELAY_HEADER_VALUE, RELAY_LAST_SLOT}, state::{BuilderApiState, PbsState}, utils::{ builder_rejection, epbs_base_send_headers, format_gwei_as_eth, log_mux_selection, @@ -81,9 +84,9 @@ pub async fn handle_get_execution_payload_bid( /// A relay's bid with the body and encoding it arrived in, so a beacon node /// that asks for the same encoding gets the relay's bytes unchanged pub(crate) struct RelayBid { - bid: GetExecutionPayloadBidResponse, - body: Bytes, - encoding: EncodingType, + pub(super) bid: GetExecutionPayloadBidResponse, + pub(super) body: Bytes, + pub(super) encoding: EncodingType, } fn encode_bid_response( @@ -168,9 +171,24 @@ pub async fn get_execution_payload_bid( let slot = params.slot; let relay_id = relay.id.clone(); + let stream_url = relay.get_execution_payload_bid_stream_url( + params.slot, + ¶ms.parent_hash, + ¶ms.parent_root, + ¶ms.proposer_pubkey, + ); + let res = match stream_url { + Some(url) => { + stream_execution_payload_bid(params, body, relay, send_headers, url, max_timeout_ms) + .await + } + None => { + send_get_execution_payload_bid(params, body, relay, send_headers, max_timeout_ms).await + } + }; // A relay that errors or times out contributes no bid: 204, never a 502. // The builder's own 400 and 401 still reach the proposer. - match send_get_execution_payload_bid(params, body, relay, send_headers, max_timeout_ms).await { + match res { Ok(Some(relay_bid)) => { RELAY_LAST_SLOT.with_label_values(&[relay_id.as_str()]).set(slot as i64); // The bid's value is already gwei, the gauge's unit, so it is set unscaled @@ -191,6 +209,39 @@ pub async fn get_execution_payload_bid( } } +/// The bid over the relay's stream. As on get_header, a stream that fails to +/// connect, a handshake answered with anything but the upgrade included, is +/// retried over HTTP with the time left; any other failure is no bid. +async fn stream_execution_payload_bid( + params: GetExecutionPayloadBidParams, + body: Bytes, + relay: RelayClient, + send_headers: HeaderMap, + url: Url, + timeout_ms: u64, +) -> Result, PbsError> { + let started = Instant::now(); + let res = get_execution_payload_bid_ws(&body, &relay, &send_headers, url, timeout_ms).await; + let timeout_left_ms = timeout_ms.saturating_sub(started.elapsed().as_millis() as u64); + match res { + Err(err @ PbsError::WebSocketConnect(_)) if timeout_left_ms > 0 => { + RELAY_EPBS_STREAM_FALLBACK.with_label_values(&[relay.id.as_str()]).inc(); + warn!( + relay_id = relay.id.as_ref(), + %err, + timeout_left_ms, + "stream failed, falling back to http" + ); + send_get_execution_payload_bid(params, body, relay, send_headers, timeout_left_ms).await + } + Err(err) => { + warn!(relay_id = relay.id.as_ref(), %err, "stream failed"); + Ok(None) + } + res => res, + } +} + /// Milliseconds left until `Date-Milliseconds + X-Timeout-Ms`, but no more than /// `X-Timeout-Ms` from now (a proposer clock running ahead) or one slot (the /// header has no upper bound). 0 once the deadline has passed. @@ -280,24 +331,58 @@ async fn send_get_execution_payload_bid( .to_string(), code: code.as_u16(), })?; - let data = - SignedExecutionPayloadBid::from_ssz_bytes(&response_bytes).map_err(|err| { - PbsError::SSZDecode { - err: format!("error decoding relay payload: {err:?}"), - fork, - } - })?; - GetExecutionPayloadBidResponse { version: fork, data, metadata: Default::default() } + decode_ssz_bid(&response_bytes, fork)? } }; Ok(Some(RelayBid { bid, body: response_bytes.into(), encoding: content_type })) } +pub(super) fn decode_ssz_bid( + bytes: &[u8], + fork: ForkName, +) -> Result { + let data = SignedExecutionPayloadBid::from_ssz_bytes(bytes).map_err(|err| { + PbsError::SSZDecode { err: format!("error decoding relay payload: {err:?}"), fork } + })?; + Ok(GetExecutionPayloadBidResponse { version: fork, data, metadata: Default::default() }) +} + #[cfg(test)] mod tests { + use std::{ + net::SocketAddr, + sync::{ + Arc, Mutex, + atomic::{AtomicU64, Ordering}, + }, + }; + + use alloy::primitives::B256; + use axum::http::Method; + use cb_common::{ + config::{GetHeaderTransport, PbsModuleConfig, RelayConfig}, + pbs::{BuilderRequestAuth, ExecutionPayloadBid, PROPOSER_DEADLINE_BUFFER_MS, RelayEntry}, + types::{BlsSecretKey, BlsSignature, Chain}, + }; + use futures::SinkExt; + use lh_types::Slot; + use prometheus::TextEncoder; + use tokio::net::{TcpListener, TcpStream}; + use tokio_tungstenite::{accept_async, tungstenite::Message}; use super::*; + use crate::{ + constants::{ + GET_EXECUTION_PAYLOAD_BID_STREAM_ENDPOINT_TAG, TIMEOUT_ERROR_STATUS, + TRANSPORT_ERROR_STATUS, + }, + metrics::{ + PBS_METRICS_REGISTRY, RELAY_EPBS_STREAM_CONNECT_LATENCY, + RELAY_EPBS_STREAM_INVALID_FRAMES, RELAY_EPBS_STREAM_UPDATES, RELAY_LATENCY, + RELAY_STATUS_CODE, + }, + }; #[test] fn test_request_budget_ms() { @@ -350,4 +435,372 @@ mod tests { )); } } + + fn relay_bid(value: u64, execution_payment: u64) -> RelayBid { + let data = SignedExecutionPayloadBid { + message: ExecutionPayloadBid { value, execution_payment, ..Default::default() }, + signature: BlsSignature::empty(), + }; + let bid = GetExecutionPayloadBidResponse { + version: ForkName::Gloas, + data, + metadata: Default::default(), + }; + RelayBid { bid, body: Bytes::new(), encoding: EncodingType::Ssz } + } + + /// Counts the relay's HTTP bid requests, told apart from a stream + /// handshake by their method + async fn count_post(socket: &TcpStream, posts: &AtomicU64) { + let mut head = [0u8; 4]; + if socket.peek(&mut head).await.is_ok_and(|n| head[..n].starts_with(b"POST")) { + posts.fetch_add(1, Ordering::Relaxed); + } + } + + /// A relay that answers every request, the stream handshake included, with + /// `status` and no body + async fn start_status_relay(status: StatusCode) -> (SocketAddr, Arc) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let posts = Arc::new(AtomicU64::new(0)); + let counted = posts.clone(); + let relay = axum::Router::new().fallback(move |method: Method| async move { + if method == Method::POST { + counted.fetch_add(1, Ordering::Relaxed); + } + status + }); + tokio::spawn(async move { axum::serve(listener, relay).await }); + (addr, posts) + } + + /// A relay whose stream sends `messages`, `gap` apart, then closes, or with + /// `close` false drops the connection with no close frame. Its HTTP request + /// fails. + async fn start_stream_relay( + messages: Vec, + gap: Duration, + close: bool, + ) -> (SocketAddr, Arc) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let posts = Arc::new(AtomicU64::new(0)); + let counted = posts.clone(); + tokio::spawn(async move { + while let Ok((socket, _)) = listener.accept().await { + let messages = messages.clone(); + let counted = counted.clone(); + tokio::spawn(async move { + count_post(&socket, &counted).await; + let Ok(mut stream) = accept_async(socket).await else { return }; + for message in messages { + let _ = stream.send(message).await; + tokio::time::sleep(gap).await; + } + if close { + let _ = stream.close(None).await; + } + }); + } + }); + (addr, posts) + } + + /// A relay that accepts the connection and never answers + async fn start_silent_relay() -> (SocketAddr, Arc) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let posts = Arc::new(AtomicU64::new(0)); + let counted = posts.clone(); + tokio::spawn(async move { + while let Ok((socket, _)) = listener.accept().await { + let counted = counted.clone(); + tokio::spawn(async move { + count_post(&socket, &counted).await; + std::future::pending::<()>().await; + }); + } + }); + (addr, posts) + } + + /// A relay that refuses the stream handshake after `delay` and answers the + /// HTTP request with a `(40, 2)` bid. Returns each request's arrival and + /// `X-Timeout-Ms`, the handshake's first. + async fn start_refusing_relay( + delay: Duration, + ) -> (SocketAddr, Arc>>) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let requests = Arc::new(Mutex::new(Vec::new())); + let recorded = requests.clone(); + let relay = axum::Router::new().fallback(move |method: Method, headers: HeaderMap| { + let recorded = recorded.clone(); + async move { + let sent = headers.get(HEADER_TIMEOUT_MS).and_then(|v| v.to_str().ok()); + let timeout_ms = sent.and_then(|v| v.parse().ok()).unwrap(); + recorded.lock().unwrap().push((Instant::now(), timeout_ms)); + if method != Method::POST { + tokio::time::sleep(delay).await; + return StatusCode::NOT_FOUND.into_response(); + } + let mut res = relay_bid(40, 2).bid.data.as_ssz_bytes().into_response(); + let headers = res.headers_mut(); + headers.insert(CONTENT_TYPE, EncodingType::Ssz.content_type_header().clone()); + headers.insert(CONSENSUS_VERSION_HEADER, GLOAS_CONSENSUS_VERSION.clone()); + res + } + }); + tokio::spawn(async move { axum::serve(listener, relay).await }); + (addr, requests) + } + + fn bid_frame(value: u64, execution_payment: u64) -> Message { + let data = relay_bid(value, execution_payment).bid.data; + // Message type bid, fork gloas + let mut frame = vec![0x01, 7]; + frame.extend(data.as_ssz_bytes()); + Message::Binary(frame.into()) + } + + /// An ePBS bid request for slot 1 addressed to one stream relay at `addr`, + /// with `budget_ms` until the beacon node's deadline. Returns the relay's + /// id, which labels its metrics, and the bid. + async fn bid_from_stream_relay(addr: SocketAddr, budget_ms: u64) -> (String, Option) { + let pubkey = BlsSecretKey::random().public_key(); + let relay_id = format!("epbs_stream_relay_{}", addr.port()); + let relay = RelayClient::new(RelayConfig { + entry: RelayEntry { + id: relay_id.clone(), + pubkey: pubkey.clone(), + url: format!("http://{pubkey}@{addr}").parse().unwrap(), + }, + id: None, + headers: None, + get_params: None, + get_header: GetHeaderTransport::Stream, + enable_timing_games: false, + target_first_request_ms: None, + frequency_get_header_ms: None, + validator_registration_batch_size: None, + }) + .unwrap(); + let state = PbsState::new( + PbsModuleConfig { + chain: Chain::Hoodi, + endpoint: addr, + pbs_config: Arc::new(serde_json::from_str("{}").unwrap()), + relays: vec![relay.clone()], + all_relays: vec![relay], + signer_client: None, + registry_muxes: None, + mux_lookup: None, + }, + Default::default(), + ); + let params = GetExecutionPayloadBidParams { + slot: 1, + parent_hash: B256::ZERO, + parent_root: B256::ZERO, + proposer_pubkey: pubkey, + }; + let auth = SignedBuilderRequestAuth { + message: BuilderRequestAuth { + data: b"127.0.0.1".to_vec().try_into().unwrap(), + slot: Slot::new(1), + }, + signature: BlsSignature::empty(), + }; + let mut req_headers = HeaderMap::new(); + req_headers.insert(HEADER_START_TIME_UNIX_MS, HeaderValue::from(utcnow_ms())); + req_headers.insert(HEADER_TIMEOUT_MS, HeaderValue::from(budget_ms)); + + let bid = get_execution_payload_bid(params, auth, req_headers, state).await.unwrap(); + (relay_id, bid) + } + + const STREAM: &str = GET_EXECUTION_PAYLOAD_BID_STREAM_ENDPOINT_TAG; + const ZERO: Duration = Duration::ZERO; + + fn status_count(status: StatusCode, relay_id: &str) -> u64 { + RELAY_STATUS_CODE.with_label_values(&[status.as_str(), STREAM, relay_id]).get() + } + + fn fallbacks(relay_id: &str) -> u64 { + RELAY_EPBS_STREAM_FALLBACK.with_label_values(&[relay_id]).get() + } + + // Only a refused handshake falls back to HTTP, a 2xx in place of the + // upgrade included: one that runs out the window, a stream that breaks + // before a bid that decodes, and a frame over the size cap are no bid, with + // no HTTP request. Only a completed handshake observes the handshake + // latency and the updates per stream. + #[tokio::test] + async fn test_epbs_stream_failures_by_status() { + let oversized = Message::Binary(vec![0; MAX_SIZE_GET_HEADER_RESPONSE + 1].into()); + let bad_bid = Message::Binary(vec![0x01, 7, 1, 2, 3].into()); + for ((relay, posts), status, fell_back, connected) in [ + (start_status_relay(StatusCode::NO_CONTENT).await, TRANSPORT_ERROR_STATUS, 1, 0), + (start_status_relay(StatusCode::NOT_FOUND).await, StatusCode::NOT_FOUND, 1, 0), + (start_status_relay(StatusCode::OK).await, TRANSPORT_ERROR_STATUS, 1, 0), + (start_silent_relay().await, TIMEOUT_ERROR_STATUS, 0, 0), + (start_stream_relay(vec![], ZERO, false).await, TRANSPORT_ERROR_STATUS, 0, 1), + (start_stream_relay(vec![bad_bid], ZERO, false).await, TRANSPORT_ERROR_STATUS, 0, 1), + (start_stream_relay(vec![oversized], ZERO, true).await, TRANSPORT_ERROR_STATUS, 0, 1), + ] { + let (relay_id, bid) = + tokio::time::timeout(Duration::from_secs(5), bid_from_stream_relay(relay, 300)) + .await + .expect("the deadline ends every stream"); + assert!(bid.is_none(), "{status}"); + assert_eq!(status_count(status, &relay_id), 1, "{status}"); + assert_eq!(fallbacks(&relay_id), fell_back, "{status}"); + assert_eq!(posts.load(Ordering::Relaxed), fell_back, "{status}"); + let relay = [relay_id.as_str()]; + let handshakes = RELAY_EPBS_STREAM_CONNECT_LATENCY.with_label_values(&relay); + assert_eq!(handshakes.get_sample_count(), connected, "{status}"); + let windows = RELAY_EPBS_STREAM_UPDATES.with_label_values(&relay); + assert_eq!(windows.get_sample_count(), connected, "{status}"); + } + } + + // The HTTP request a refused handshake falls back to has the time the + // handshake left, and its bid is served + #[tokio::test] + async fn test_epbs_stream_refused_handshake_falls_back_with_the_time_left() { + let (relay, requests) = start_refusing_relay(Duration::from_millis(200)).await; + let (relay_id, bid) = bid_from_stream_relay(relay, 1_000).await; + let bid = bid.expect("the HTTP bid").bid; + assert_eq!((bid.data.message.value, bid.data.message.execution_payment), (40, 2)); + assert_eq!(status_count(StatusCode::NOT_FOUND, &relay_id), 1); + assert_eq!(fallbacks(&relay_id), 1); + + // 1000 ms less the default buffer and the 200 ms handshake + let [handshake, fallback] = requests.lock().unwrap()[..] else { panic!("two requests") }; + assert!((1..=800 - PROPOSER_DEADLINE_BUFFER_MS).contains(&fallback.1), "{}", fallback.1); + + // Both requests name the same deadline: the fallback's time left is + // the handshake's, less the time the handshake took + let deadline = |(at, timeout_ms): (Instant, u64)| at + Duration::from_millis(timeout_ms); + let (handshake, fallback) = (deadline(handshake), deadline(fallback)); + let drift = fallback.max(handshake) - fallback.min(handshake); + assert!(drift < Duration::from_millis(100), "{drift:?}"); + } + + // A stream that breaks after a bid arrived serves that bid, with no HTTP + // request + #[tokio::test] + async fn test_epbs_stream_break_after_a_bid_serves_it() { + let (relay, posts) = start_stream_relay(vec![bid_frame(9, 1)], ZERO, false).await; + let (relay_id, bid) = bid_from_stream_relay(relay, 1_000).await; + let bid = bid.expect("the bid before the break").bid; + assert_eq!((bid.data.message.value, bid.data.message.execution_payment), (9, 1)); + assert_eq!(status_count(StatusCode::OK, &relay_id), 1); + assert_eq!((fallbacks(&relay_id), posts.load(Ordering::Relaxed)), (0, 0)); + } + + // A bid for a fork other than Gloas is an invalid frame, and leaves the + // bid before it served + #[tokio::test] + async fn test_epbs_stream_wrong_fork_bid_is_invalid() { + let mut fulu = bid_frame(90, 0).into_data().to_vec(); + fulu[1] = 6; + let messages = vec![bid_frame(9, 1), Message::Binary(fulu.into())]; + let (relay, _) = start_stream_relay(messages, ZERO, true).await; + let (relay_id, bid) = bid_from_stream_relay(relay, 1_000).await; + let bid = bid.expect("the gloas bid").bid; + assert_eq!((bid.version, bid.data.message.value), (ForkName::Gloas, 9)); + let invalid = RELAY_EPBS_STREAM_INVALID_FRAMES.with_label_values(&[relay_id.as_str()]); + assert_eq!(invalid.get(), 1); + } + + // A window whose every bid fails to decode is no bid, not a failed stream, + // and counts each of those bids as an invalid frame + #[tokio::test] + async fn test_epbs_stream_undecodable_bids_are_no_bid() { + for n_frames in [1, 3] { + let undecodable = Message::Binary(vec![0x01, 7, 1, 2, 3].into()); + let (relay, posts) = start_stream_relay(vec![undecodable; n_frames], ZERO, true).await; + let (relay_id, bid) = bid_from_stream_relay(relay, 1_000).await; + assert!(bid.is_none()); + + assert_eq!(status_count(StatusCode::NO_CONTENT, &relay_id), 1); + assert_eq!(fallbacks(&relay_id), 0); + assert_eq!(posts.load(Ordering::Relaxed), 0); + let relay = [relay_id.as_str()]; + assert_eq!( + RELAY_EPBS_STREAM_INVALID_FRAMES.with_label_values(&relay).get(), + n_frames as u64 + ); + assert_eq!(RELAY_EPBS_STREAM_UPDATES.with_label_values(&relay).get_sample_sum(), 0.0); + // No bid, so no time to first bid + let latency = RELAY_LATENCY.with_label_values(&[STREAM, relay_id.as_str()]); + assert_eq!(latency.get_sample_count(), 0); + } + } + + // Messages other than bids do not end the stream. The served bid keeps the + // frame's fork and sets the relay's gauges: its value without the execution + // payment, and the slot. The latency series times the first bid. The relay + // gets no HTTP request. + #[tokio::test] + async fn test_epbs_stream_bid_after_other_messages_is_served_and_recorded() { + let messages = vec![ + bid_frame(30, 0), + Message::Text("hello".into()), + Message::Ping(vec![1].into()), + bid_frame(77, 5), + ]; + let (relay, posts) = start_stream_relay(messages, Duration::from_millis(150), true).await; + let start = Instant::now(); + let (relay_id, bid) = bid_from_stream_relay(relay, 2_000).await; + // The relay closes 600 ms in, long before the 1950 ms deadline + assert!(start.elapsed() < Duration::from_millis(1_500), "{:?}", start.elapsed()); + let bid = bid.expect("the stream's bid").bid; + assert_eq!(bid.version, ForkName::Gloas); + let bid = &bid.data.message; + assert_eq!((bid.value, bid.execution_payment), (77, 5)); + assert_eq!(posts.load(Ordering::Relaxed), 0); + + assert_eq!(status_count(StatusCode::OK, &relay_id), 1); + assert_eq!(RELAY_HEADER_VALUE.with_label_values(&[&relay_id]).get(), 77); + assert_eq!(RELAY_LAST_SLOT.with_label_values(&[&relay_id]).get(), 1); + let relay = [relay_id.as_str()]; + assert_eq!(RELAY_EPBS_STREAM_UPDATES.with_label_values(&relay).get_sample_sum(), 2.0); + assert_eq!( + RELAY_EPBS_STREAM_CONNECT_LATENCY.with_label_values(&relay).get_sample_count(), + 1 + ); + // The last bid arrives 450 ms in + let latency = RELAY_LATENCY.with_label_values(&[STREAM, relay_id.as_str()]); + assert_eq!(latency.get_sample_count(), 1); + assert!(latency.get_sample_sum() < 0.3, "{}", latency.get_sample_sum()); + + // No frame was invalid, so that series is absent rather than zero + let scrape = TextEncoder::new().encode_to_string(&PBS_METRICS_REGISTRY.gather()).unwrap(); + let invalid = format!(r#"relay_epbs_stream_invalid_frames_total{{relay_id="{relay_id}"}}"#); + assert!(!scrape.contains(&invalid), "{invalid}"); + } + + // Dashboards and the docs select the stream series by name + #[tokio::test] + async fn test_epbs_stream_series_are_labeled_by_name() { + let messages = vec![Message::Binary(vec![0x01, 7, 1].into()), bid_frame(1, 0)]; + let (relay_id, _) = + bid_from_stream_relay(start_stream_relay(messages, ZERO, true).await.0, 1_000).await; + let (failed_id, _) = + bid_from_stream_relay(start_status_relay(StatusCode::NOT_FOUND).await.0, 300).await; + + let scrape = TextEncoder::new().encode_to_string(&PBS_METRICS_REGISTRY.gather()).unwrap(); + for (series, relay_id) in [ + ("relay_epbs_stream_connect_latency_count", &relay_id), + ("relay_epbs_stream_updates_sum", &relay_id), + ("relay_epbs_stream_invalid_frames_total", &relay_id), + ("relay_epbs_stream_fallback_total", &failed_id), + ] { + let series = format!(r#"{series}{{relay_id="{relay_id}"}}"#); + assert!(scrape.contains(&series), "{series}"); + } + } } diff --git a/crates/pbs/src/routes/execution_payload_bid_ws.rs b/crates/pbs/src/routes/execution_payload_bid_ws.rs new file mode 100644 index 00000000..937f2409 --- /dev/null +++ b/crates/pbs/src/routes/execution_payload_bid_ws.rs @@ -0,0 +1,406 @@ +//! The ePBS bid over a websocket, for relays configured with +//! `get_header = "stream"`. The stream URL is derived from the relay's own +//! `url`. One connection per bid request, dropped when the request returns. +//! +//! The request is the handshake itself: slot / parent_hash / parent_root / +//! pubkey in the path, deadline, timestamp and request auth in headers, same +//! data the HTTP request carries. The relay replies with one binary frame per +//! bid update: +//! +//! ```text +//! u8 message type +//! u8 fork +//! .. SSZ SignedExecutionPayloadBid +//! ``` + +use std::{ + sync::{Arc, OnceLock}, + time::Duration, +}; + +use axum::http::{HeaderMap, HeaderValue, Request, header::USER_AGENT}; +use base64::{Engine, prelude::BASE64_STANDARD}; +use cb_common::{ + pbs::{ + ForkName, HEADER_REQUEST_AUTH, HEADER_START_TIME_UNIX_MS, HEADER_TIMEOUT_MS, RelayClient, + error::PbsError, + }, + utils::utcnow_ms, + wire::{CONSENSUS_VERSION_HEADER, EncodingType, GLOAS_CONSENSUS_VERSION}, +}; +use futures::StreamExt; +use reqwest::StatusCode; +use rustls::{ClientConfig, RootCertStore, crypto::aws_lc_rs}; +use tokio::time::{Instant, sleep_until, timeout_at}; +use tokio_tungstenite::{ + Connector, connect_async_tls_with_config, + tungstenite::{ + Bytes, Error as WsError, Message, client::IntoClientRequest, protocol::WebSocketConfig, + }, +}; +use tracing::{debug, warn}; +use url::Url; + +use super::execution_payload_bid::{RelayBid, decode_ssz_bid}; +use crate::{ + constants::{ + GET_EXECUTION_PAYLOAD_BID_STREAM_ENDPOINT_TAG, MAX_SIZE_GET_HEADER_RESPONSE, + TIMEOUT_ERROR_STATUS, TRANSPORT_ERROR_STATUS, + }, + metrics::{ + RELAY_EPBS_STREAM_CONNECT_LATENCY, RELAY_EPBS_STREAM_INVALID_FRAMES, + RELAY_EPBS_STREAM_UPDATES, RELAY_LATENCY, RELAY_STATUS_CODE, + }, +}; + +/// Frame prefix: message type + fork. +const FRAME_PREFIX_LEN: usize = 2; + +const MSG_BID: u8 = 0x01; + +fn fork_from_wire(byte: u8) -> Option { + Some(match byte { + 0 => ForkName::Base, + 1 => ForkName::Altair, + 2 => ForkName::Bellatrix, + 3 => ForkName::Capella, + 4 => ForkName::Deneb, + 5 => ForkName::Electra, + 6 => ForkName::Fulu, + 7 => ForkName::Gloas, + 8 => ForkName::Heze, + _ => return None, + }) +} + +type StreamOutcome = (StatusCode, Result, PbsError>); + +/// Open a stream to the relay and return its latest bid at the deadline +pub(super) async fn get_execution_payload_bid_ws( + auth_ssz: &[u8], + relay: &RelayClient, + send_headers: &HeaderMap, + url: Url, + timeout_ms: u64, +) -> Result, PbsError> { + let (status, res) = stream_bid(auth_ssz, relay, send_headers, url, timeout_ms).await; + let endpoint = GET_EXECUTION_PAYLOAD_BID_STREAM_ENDPOINT_TAG; + RELAY_STATUS_CODE.with_label_values(&[status.as_str(), endpoint, &relay.id]).inc(); + res +} + +async fn stream_bid( + auth_ssz: &[u8], + relay: &RelayClient, + send_headers: &HeaderMap, + url: Url, + timeout_ms: u64, +) -> StreamOutcome { + let deadline = Instant::now() + Duration::from_millis(timeout_ms); + let request = match build_handshake_request(auth_ssz, relay, send_headers, &url, timeout_ms) { + Ok(request) => request, + Err(err) => return (TRANSPORT_ERROR_STATUS, Err(err)), + }; + + let config = WebSocketConfig::default() + .max_message_size(Some(MAX_SIZE_GET_HEADER_RESPONSE)) + .max_frame_size(Some(MAX_SIZE_GET_HEADER_RESPONSE)); + + let start_request = Instant::now(); + let connect = connect_async_tls_with_config( + request, + Some(config), + true, + Some(Connector::Rustls(tls_config().clone())), + ); + let (mut stream, _) = match timeout_at(deadline, connect).await { + Ok(Ok(connected)) => connected, + Ok(Err(err)) => return connect_failed(&err), + Err(_) => { + return (TIMEOUT_ERROR_STATUS, Err(PbsError::WebSocketTimeout)); + } + }; + let connect_latency = start_request.elapsed(); + RELAY_EPBS_STREAM_CONNECT_LATENCY + .with_label_values(&[relay.id.as_str()]) + .observe(connect_latency.as_secs_f64()); + debug!(relay_id = relay.id.as_ref(), ?connect_latency, "ws connected"); + + let timer = sleep_until(deadline); + tokio::pin!(timer); + + let mut latest: Option = None; + let mut first_bid_latency: Option = None; + let mut updates = 0usize; + let mut invalid_frames = 0usize; + let mut stream_error = None; + + loop { + let message = tokio::select! { + biased; + _ = &mut timer => break, + message = stream.next() => message, + }; + + let message = match message { + Some(Ok(message)) => message, + Some(Err(err)) => { + warn!(relay_id = relay.id.as_ref(), %err, "ws stream error"); + stream_error = Some(PbsError::WebSocket(format!("stream error: {err}"))); + break; + } + None => break, + }; + + let payload = match message { + Message::Binary(payload) => payload, + Message::Close(_) => break, + _ => continue, + }; + + match parse_frame(payload).and_then(decode_bid) { + Ok(bid) => { + updates += 1; + if first_bid_latency.is_none() { + first_bid_latency = Some(start_request.elapsed()); + } + latest = Some(bid); + } + Err(err) => { + invalid_frames += 1; + if invalid_frames == 1 { + warn!(relay_id = relay.id.as_ref(), %err, "invalid ws frame, skipping"); + } + } + } + } + + drop(stream); + + RELAY_EPBS_STREAM_UPDATES.with_label_values(&[relay.id.as_str()]).observe(updates as f64); + if invalid_frames > 0 { + RELAY_EPBS_STREAM_INVALID_FRAMES + .with_label_values(&[relay.id.as_str()]) + .inc_by(invalid_frames as u64); + } + + let Some(bid) = latest else { + if let Some(err) = stream_error { + return (TRANSPORT_ERROR_STATUS, Err(err)); + } + + debug!(relay_id = relay.id.as_ref(), ?connect_latency, invalid_frames, "no bid"); + return (StatusCode::NO_CONTENT, Ok(None)); + }; + + if let Some(first_bid_latency) = first_bid_latency { + RELAY_LATENCY + .with_label_values(&[GET_EXECUTION_PAYLOAD_BID_STREAM_ENDPOINT_TAG, &relay.id]) + .observe(first_bid_latency.as_secs_f64()); + } + + debug!( + relay_id = relay.id.as_ref(), + ?connect_latency, + ?first_bid_latency, + updates, + invalid_frames, + "received new bid from ws stream" + ); + + (StatusCode::OK, Ok(Some(bid))) +} + +/// A frame's bid, which must be for Gloas +fn decode_bid((fork, body): (ForkName, Bytes)) -> Result { + if fork != ForkName::Gloas { + return Err(PbsError::WebSocket(format!("bid for {fork}, expected gloas"))); + } + let bid = decode_ssz_bid(&body, fork)?; + Ok(RelayBid { bid, body, encoding: EncodingType::Ssz }) +} + +/// A relay rejecting the handshake puts the reason in the body, which +/// tungstenite's own `Display` drops. The body is only what arrived alongside +/// the headers, so it can be partial or empty. +fn connect_failed(err: &WsError) -> StreamOutcome { + let WsError::Http(res) = err else { + return (TRANSPORT_ERROR_STATUS, Err(PbsError::WebSocketConnect(err.to_string()))); + }; + + let code = res.status(); + let body = res.body().as_deref().unwrap_or_default(); + let msg = if body.is_empty() { + format!("rejected with {code}") + } else { + format!("rejected with {code}: {}", String::from_utf8_lossy(body)) + }; + + // A 2xx handshake answer is a failed connect, not a delivered bid + let code = if code.is_success() { TRANSPORT_ERROR_STATUS } else { code }; + + (code, Err(PbsError::WebSocketConnect(msg))) +} + +fn build_handshake_request( + auth_ssz: &[u8], + relay: &RelayClient, + send_headers: &HeaderMap, + url: &Url, + timeout_ms: u64, +) -> Result, PbsError> { + let mut request = url + .as_str() + .into_client_request() + .map_err(|err| PbsError::WebSocketConnect(format!("invalid ws url: {err}")))?; + + let headers = request.headers_mut(); + if let Some(user_agent) = send_headers.get(USER_AGENT) { + headers.insert(USER_AGENT, user_agent.clone()); + } + + for (key, value) in relay.stream_headers() { + headers.insert(key, value.clone()); + } + + headers.insert(CONSENSUS_VERSION_HEADER, GLOAS_CONSENSUS_VERSION.clone()); + let mut request_auth = HeaderValue::try_from(BASE64_STANDARD.encode(auth_ssz)) + .expect("base64 is always a valid header value"); + request_auth.set_sensitive(true); + headers.insert(HEADER_REQUEST_AUTH, request_auth); + + headers.insert(HEADER_START_TIME_UNIX_MS, HeaderValue::from(utcnow_ms())); + headers.insert(HEADER_TIMEOUT_MS, HeaderValue::from(timeout_ms)); + + Ok(request) +} + +fn parse_frame(payload: Bytes) -> Result<(ForkName, Bytes), PbsError> { + let &[msg_type, fork_byte] = payload + .first_chunk::() + .ok_or_else(|| PbsError::WebSocket(format!("frame too short: {} bytes", payload.len())))?; + + if msg_type != MSG_BID { + return Err(PbsError::WebSocket(format!("unknown message type: {msg_type}"))); + } + + let fork = fork_from_wire(fork_byte) + .ok_or_else(|| PbsError::WebSocket(format!("unknown fork: {fork_byte}")))?; + + Ok((fork, payload.slice(FRAME_PREFIX_LEN..))) +} + +/// One TLS config for every ePBS stream connection. Left to tokio-tungstenite +/// it is rebuilt per connect, which reparses the root store and, worse, gives +/// each connection its own session cache: every slot then pays a full handshake +/// instead of a resumed one. The provider is named explicitly because rustls is +/// built with both `ring` and `aws-lc-rs` here, so the default builder needs a +/// process-wide install to pick one. +fn tls_config() -> &'static Arc { + static CONFIG: OnceLock> = OnceLock::new(); + CONFIG.get_or_init(|| { + let mut roots = RootCertStore::empty(); + roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned()); + + Arc::new( + ClientConfig::builder_with_provider(Arc::new(aws_lc_rs::default_provider())) + .with_safe_default_protocol_versions() + .expect("aws-lc-rs supports tls 1.2 and 1.3") + .with_root_certificates(roots) + .with_no_client_auth(), + ) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn bid_frame(fork_byte: u8, bid: &[u8]) -> Bytes { + let mut frame = vec![MSG_BID, fork_byte]; + frame.extend_from_slice(bid); + Bytes::from(frame) + } + + #[test] + fn test_fork_from_wire_covers_all_forks() { + let forks = ForkName::list_all(); + + for (byte, fork) in forks.iter().enumerate() { + if *fork <= ForkName::Base { + continue; + } + assert_eq!(fork_from_wire(byte as u8), Some(*fork), "fork {fork} unmapped"); + } + + assert_eq!(fork_from_wire(forks.len() as u8), None); + } + + #[test] + fn test_parse_frame() { + assert!(matches!( + parse_frame(bid_frame(6, &[1, 2, 3])), + Ok((ForkName::Fulu, bid)) if bid.as_ref() == [1, 2, 3] + )); + + // Empty bid payload is well-formed at this layer, SSZ decoding rejects it + assert!(matches!(parse_frame(bid_frame(6, &[])), Ok((ForkName::Fulu, _)))); + + for bad in [ + // Truncated prefix + Bytes::from_static(&[]), + Bytes::from_static(&[MSG_BID]), + // Unknown fork + bid_frame(0xff, &[1]), + // Unknown message type + Bytes::from_static(&[0xff, 6]), + ] { + assert!(matches!(parse_frame(bad), Err(PbsError::WebSocket(_)))); + } + } + + #[test] + fn test_connect_failed_carries_relay_reason() { + let rejected = axum::http::Response::builder() + .status(401) + .body(Some(b"api key not registered".to_vec())) + .unwrap(); + + let (status, res) = connect_failed(&WsError::Http(Box::new(rejected))); + assert_eq!(status, StatusCode::UNAUTHORIZED); + let Err(PbsError::WebSocketConnect(msg)) = res else { panic!("wrong outcome") }; + assert!(msg.contains("401"), "{msg}"); + assert!(msg.contains("api key not registered"), "{msg}"); + + // No body to add, and transport failures keep tungstenite's own message + let empty = axum::http::Response::builder().status(404).body(None).unwrap(); + let (status, _) = connect_failed(&WsError::Http(Box::new(empty))); + assert_eq!(status, StatusCode::NOT_FOUND); + + let (status, res) = connect_failed(&WsError::ConnectionClosed); + assert_eq!(status, TRANSPORT_ERROR_STATUS); + assert!(matches!(res, Err(PbsError::WebSocketConnect(_)))); + } + + // A url pointing at a plain http endpoint answers the handshake 200. That + // is the code the stream series uses for a delivered bid, so a failed + // handshake must never carry it. + #[test] + fn test_connect_failed_never_reports_a_success_code() { + for code in [200u16, 204, 299] { + let answered = axum::http::Response::builder().status(code).body(None).unwrap(); + let (status, res) = connect_failed(&WsError::Http(Box::new(answered))); + assert_eq!( + status, TRANSPORT_ERROR_STATUS, + "handshake answered {code} counted as a served stream" + ); + let Err(PbsError::WebSocketConnect(msg)) = res else { panic!("wrong outcome") }; + assert!(msg.contains(&code.to_string()), "{msg}"); + } + + // A relay's own rejection code still reaches the series unchanged + let moved = axum::http::Response::builder().status(302).body(None).unwrap(); + let (status, _) = connect_failed(&WsError::Http(Box::new(moved))); + assert_eq!(status, StatusCode::FOUND); + } +} diff --git a/crates/pbs/src/routes/mod.rs b/crates/pbs/src/routes/mod.rs index 595552f5..174b5d08 100644 --- a/crates/pbs/src/routes/mod.rs +++ b/crates/pbs/src/routes/mod.rs @@ -1,5 +1,6 @@ mod builder_preferences; mod execution_payload_bid; +mod execution_payload_bid_ws; mod get_header; mod register_validator; mod reload; diff --git a/docs/docs/get_started/configuration.md b/docs/docs/get_started/configuration.md index 075c61e7..55c6314e 100644 --- a/docs/docs/get_started/configuration.md +++ b/docs/docs/get_started/configuration.md @@ -112,6 +112,8 @@ Under Docker, `commit-boost init` mounts each `file` path into the container at `get_header = "stream"` replaces the per-slot HTTP request with a websocket. It is set per relay, on `[[relays]]` and `[[mux.relays]]` entries alike. [`examples/configs/pbs_bid_stream.toml`](https://github.com/Commit-Boost/commit-boost-client/blob/main/examples/configs/pbs_bid_stream.toml) streams from two default relays and two mux relays. +(unreleased, from v0.12.0-rc1) After the Gloas fork the same setting also streams [ePBS bids](./epbs.md#bid-streaming). + #### Connection For each `get_header` call PBS dials: @@ -140,7 +142,7 @@ If the handshake fails, PBS falls back to a plain HTTP `get_header` at the relay There is no fallback if the handshake is still unanswered when the window ends (recorded as `555`), or if it fails too late to leave time for an HTTP request. There is also none once the stream is open. If it breaks before any bid arrives, that relay contributes no header for the slot (recorded as `556`). If a bid already arrived, PBS validates and returns it as usual. -On a high-latency connection, where the handshake risks timing out, you can add the relay a second time with `get_header = "http"` and a different `id`. PBS asks both entries every slot and keeps the better bid, so the HTTP entry still returns a header when the stream does not. +On a high-latency connection, where the handshake risks timing out, you can add the relay a second time with `get_header = "http"` and a different `id`. PBS asks both entries every slot and keeps the better bid, so the HTTP entry still returns a header when the stream does not. (unreleased, from v0.12.0-rc1) After the Gloas fork only the first of the two entries serves [ePBS bids](./epbs.md#bid-streaming). ```toml [[relays]] diff --git a/docs/docs/get_started/epbs.md b/docs/docs/get_started/epbs.md index 0f04c4f9..92ca6ac5 100644 --- a/docs/docs/get_started/epbs.md +++ b/docs/docs/get_started/epbs.md @@ -27,7 +27,7 @@ The ePBS bid endpoint does not use these PBS options, which will be deprecated a - `skip_sigverify`, `min_bid_eth` and `extra_validation_enabled`: the beacon node checks the bid against the on-chain builder registry and applies the `min_bid` from its builder config. - `timeout_get_header_ms`, `late_in_slot_time_ms`, their `[[mux]]` overrides and the timing-games options: the beacon node's deadline bounds the request (see [Timing](#timing)). -A `get_header = "stream"` relay is asked for ePBS bids over plain HTTP. +For a builder whose relay entry sets `get_header = "stream"`, Commit-Boost reads its bids from a [stream](#bid-streaming). ## How a bid request reaches a builder @@ -92,15 +92,28 @@ Think of `proposer_deadline_buffer_ms` as the slack you keep from the beacon nod Builder preferences use `timeout_register_validator_ms`, and the signed block `timeout_get_payload_ms`. +## Bid streaming + +A builder answers an HTTP bid request once, so a better bid it makes after answering never reaches the beacon node. Over a websocket stream it can keep sending bids until just before the beacon node's deadline. + +Ask the builder whether it serves the ePBS bid stream, at `ws(s):///eth/v1/builder/execution_payload_bid_stream/{slot}/{parent_hash}/{parent_root}/{proposer_pubkey}`. If it does, set `get_header = "stream"` on its relay entry. Before the fork the same setting reads the builder's `get_header` bids from its [PBS stream](./configuration.md#bid-streaming), so for a builder that serves only one of the two, Commit-Boost logs a failed stream on each request to the other and asks over HTTP instead. The handshake carries the relay entry's `headers`, so a builder that requires an [API key](./configuration.md#relay-api-keys) gets it. + +For each bid request to a streaming builder, Commit-Boost opens the stream and returns the latest streamed bid that decodes. Only a stream that fails to connect, because of a connection error or a handshake answered with anything but the upgrade, `204` included, is retried over HTTP with the time left. A handshake still unanswered at the deadline, or a stream that breaks before a bid, is no bid. A builder [outside your config](#builders-outside-your-config) is always asked over HTTP. If a builder has two relay entries, one on `"stream"` and one on `"http"`, only the first that matches the key's auth data serves ePBS bids. + +Streaming makes bid requests slower. Commit-Boost replies to the beacon node once the stream has ended: when the builder closes it, when it fails, or at the beacon node's deadline less `proposer_deadline_buffer_ms`. After a retry over HTTP, it replies once that request answers. A bid request to an HTTP builder is answered as soon as the builder responds. `timeout_get_header_ms` does not shorten the wait. If the beacon node reports bid requests timing out on streaming builders, raise `proposer_deadline_buffer_ms`. + +The stream connects directly, ignoring `HTTP_PROXY`, `HTTPS_PROXY` and `ALL_PROXY`, while the HTTP fallback uses them. If your host reaches builders only through a proxy, keep them on `"http"`. + ## Metrics -With [metrics](./running/metrics.md) enabled, the ePBS endpoints use the `endpoint` labels `get_execution_payload_bid`, `submit_builder_preferences` and `submit_signed_beacon_block`. +With [metrics](./running/metrics.md) enabled, the ePBS endpoints use the `endpoint` labels `get_execution_payload_bid`, `submit_builder_preferences` and `submit_signed_beacon_block`, and a [streaming](#bid-streaming) builder's stream uses `get_execution_payload_bid_stream`, with its own `cb_pbs_relay_epbs_stream_*` series ([Metrics > Bid stream](./running/metrics.md#bid-stream)). | Question | Series | |---|---| | What did Commit-Boost answer the beacon node? | `cb_pbs_beacon_node_status_code_total`: `200` or `204` for a bid request, `202` for preferences, `202` for the signed block whatever the builders answer, `4xx` for a rejected request, `500` when no builder accepted the preferences or on an internal error | -| What did each builder answer? | `cb_pbs_relay_status_code_total`, by `relay_id`: the builder's HTTP status, or `555` when no response arrived (timeout, DNS or connection failure). Requests to builders outside your config count under `relay_id="dial"`; a dial Commit-Boost [refuses](#builders-outside-your-config), or whose lookup fails, times out or finds no free slot, sends no request and is not counted | +| What did each builder answer? | `cb_pbs_relay_status_code_total`, by `relay_id`: the builder's HTTP status, or `555` when no response arrived (timeout, DNS or connection failure; on the bid stream, only the deadline). Requests to builders outside your config count under `relay_id="dial"`; a dial Commit-Boost [refuses](#builders-outside-your-config), or whose lookup fails, times out or finds no free slot, sends no request and is not counted | | How fast are builders? | `cb_pbs_relay_latency`, by `relay_id` | +| Is a builder's bid stream working? | `cb_pbs_relay_status_code_total{endpoint="get_execution_payload_bid_stream"}`: `200` when the stream supplied a bid, `204` when it had none, `555` when the deadline ran out before the handshake completed, `556` when the connection failed, the handshake was answered with a success code instead of the upgrade, or the stream dropped before a bid, or the builder's own refusal code such as `404` | | What are builders bidding? | `cb_pbs_relay_header_value` (the bid's `value` in Gwei, without the execution payment) and `cb_pbs_relay_last_slot`, by `relay_id`, from each bid a builder serves | ## Troubleshooting @@ -110,5 +123,7 @@ With [metrics](./running/metrics.md) enabled, the ePBS endpoints use the `endpoi | Bid requests get `400` "auth.message.data does not match any configured builder" | The auth data is neither a lowercase hostname without a port nor an `http(s)` URL, names the host the request was sent to (a key without builder config), or the request came from another Commit-Boost and matches none of this one's relay entries | Correct the `auth_data` in the key's [builder config](#validator-builder-config), or add the builder as a relay entry on the Commit-Boost the request reaches | | Bid requests get `400` "the addressed builder's host does not resolve or resolves to a disallowed address" | The auth data names a builder outside your config whose host does not resolve or resolves to an [internal address](#builders-outside-your-config) | Add the builder as a relay entry for the key, or write or correct the key's [builder config](#validator-builder-config) | | A builder answers bid requests with `400` (`cb_pbs_relay_status_code_total{endpoint="get_execution_payload_bid",http_status_code="400"}`) | The builder compares auth data byte for byte and expects something other than what the key sends, such as its hostname without `?` parameters | Send the auth data the builder expects: its hostname or its URL, the forms Commit-Boost [routes by](#routing-by-auth-data), with `?` parameters only if it accepts them | +| Bid requests to a streaming builder log "stream failed" with "rejected with 404", and `cb_pbs_relay_epbs_stream_fallback_total` rises | The builder does not serve the ePBS bid stream. Commit-Boost asks it over HTTP instead | Set `get_header = "http"` on its relay entry | | Commit-Boost does not start or reload, or `commit-boost init` fails, with "proposer_deadline_buffer_ms must be greater than 0 and less than one slot" | `proposer_deadline_buffer_ms` is `0`, or one slot or more | Set it above `0` and under one slot, or remove it to use the default `50` | +| Bid requests to a streaming builder log "stream failed" with "websocket timed out" or a connection error | Commit-Boost could not open the stream, for example because your network allows only proxied connections, which the stream does not use. A connection error with time left is retried over HTTP; a handshake that times out is no bid | Set `get_header = "http"` on its relay entry, or allow direct connections to it | | Commit-Boost returns `200` but the beacon node builds locally | The beacon node rejected the bid, or valued its local block higher after `builder_boost_factor` | Compare the bid with the key's `min_bid` and `builder_boost_factor` | diff --git a/docs/docs/get_started/running/metrics-catalog.md b/docs/docs/get_started/running/metrics-catalog.md index 74aba597..c096d5c3 100644 --- a/docs/docs/get_started/running/metrics-catalog.md +++ b/docs/docs/get_started/running/metrics-catalog.md @@ -14,14 +14,18 @@ PBS metrics use a custom Prometheus registry with namespace prefix `cb_pbs_`. Th | Metric name (wire) | Type | Labels | Description | |---|---|---|---| -| `cb_pbs_relay_status_code_total` | Counter | `http_status_code`, `endpoint`, `relay_id` | HTTP status code received by relay. Incremented once per relay response. Two synthetic codes stand in for outcomes that never reach a status line: `"555"` when no response arrived (timeout, DNS or connection failure; on `get_header_stream`, only the deadline), and `"556"` for a bid-stream transport failure (connect failed, the stream broke mid-window, or the handshake was answered with a success code instead of the upgrade). Endpoint values: `get_header`, `get_header_stream`, `register_validator`, `submit_blinded_block`, `status`, and (unreleased, from v0.12.0-rc1) `get_execution_payload_bid`, `submit_builder_preferences`, `submit_signed_beacon_block`. | -| `cb_pbs_relay_latency` | Histogram | `endpoint`, `relay_id` | HTTP latency (duration in seconds) by relay. Records the duration of relay HTTP requests. Endpoint values: `get_header`, `get_header_stream`, `register_validator`, `submit_blinded_block`, `status`, and (unreleased, from v0.12.0-rc1) `get_execution_payload_bid`, `submit_builder_preferences`, `submit_signed_beacon_block`. Under `get_header_stream` the observation is the time from opening the websocket to the first bid update, not a round trip, and a window that received no bid records nothing. | +| `cb_pbs_relay_status_code_total` | Counter | `http_status_code`, `endpoint`, `relay_id` | HTTP status code received by relay. Incremented once per relay response. Two synthetic codes stand in for outcomes that never reach a status line: `"555"` when no response arrived (timeout, DNS or connection failure; on `get_header_stream`, only the deadline), and `"556"` for a bid-stream transport failure (connect failed, the stream broke mid-window, or the handshake was answered with a success code instead of the upgrade). (unreleased, from v0.12.0-rc1) On `get_execution_payload_bid_stream`, as on `get_header_stream`, `555` is only the deadline. Endpoint values: `get_header`, `get_header_stream`, `register_validator`, `submit_blinded_block`, `status`, and (unreleased, from v0.12.0-rc1) `get_execution_payload_bid`, `get_execution_payload_bid_stream`, `submit_builder_preferences`, `submit_signed_beacon_block`. | +| `cb_pbs_relay_latency` | Histogram | `endpoint`, `relay_id` | HTTP latency (duration in seconds) by relay. Records the duration of relay HTTP requests. Endpoint values: `get_header`, `get_header_stream`, `register_validator`, `submit_blinded_block`, `status`, and (unreleased, from v0.12.0-rc1) `get_execution_payload_bid`, `get_execution_payload_bid_stream`, `submit_builder_preferences`, `submit_signed_beacon_block`. Under `get_header_stream` and `get_execution_payload_bid_stream` the observation is the time from opening the websocket to the first bid update, not a round trip, and a window that received no bid records nothing. | | `cb_pbs_relay_last_slot` | Gauge | `relay_id` | Latest slot for which a relay delivered a bid. Set to the current slot on each successful `get_header` bid and (unreleased, from v0.12.0-rc1) each ePBS bid served from that relay. | | `cb_pbs_relay_header_value` | Gauge | `relay_id` | Value in gwei of the bid a relay delivered: for `get_header`, the bid's value converted from wei (รท 1e9). (unreleased, from v0.12.0-rc1) For ePBS, the returned bid's `value`, without its execution payment. | | `cb_pbs_relay_stream_connect_latency` | Histogram | `relay_id` | Websocket handshake latency in seconds, for a relay with `get_header = "stream"`. Observed once per successful handshake; a handshake that fails or runs out the bid window records nothing here. Custom buckets from 5 ms to 1 s. | | `cb_pbs_relay_stream_updates` | Histogram | `relay_id` | Bid updates received per stream window. Observed once per `get_header` served by the stream, including windows that received nothing. Buckets: `0, 1, 2, 3, 5, 10, 20, 50, 100`. | | `cb_pbs_relay_stream_invalid_frames_total` | Counter | `relay_id` | Websocket frames that could not be parsed as a bid. Incremented only for windows that saw at least one, so the series is absent while zero. | | `cb_pbs_relay_stream_fallback_total` | Counter | `relay_id` | Stream attempts that fell back to a plain HTTP `get_header`. Only a handshake failure with bid window left to retry counts here; one that consumed the window shows on the status series alone. The fallback's own result is recorded under `endpoint="get_header"`. | +| `cb_pbs_relay_epbs_stream_connect_latency` | Histogram | `relay_id` | (unreleased, from v0.12.0-rc1) ePBS bid stream handshake latency in seconds. Observed once per successful handshake; a handshake that fails or runs out the deadline records nothing here. Custom buckets from 5 ms to 1 s. | +| `cb_pbs_relay_epbs_stream_updates` | Histogram | `relay_id` | (unreleased, from v0.12.0-rc1) Bids that decoded per ePBS bid stream. Observed once per stream that connected, including streams that received nothing. Buckets: `0, 1, 2, 3, 5, 10, 20, 50, 100`. | +| `cb_pbs_relay_epbs_stream_invalid_frames_total` | Counter | `relay_id` | (unreleased, from v0.12.0-rc1) ePBS bid stream frames whose message type or fork byte cannot be read, or whose bid does not decode. Incremented only for streams that saw at least one, so the series is absent while zero. | +| `cb_pbs_relay_epbs_stream_fallback_total` | Counter | `relay_id` | (unreleased, from v0.12.0-rc1) ePBS bid streams that fell back to an HTTP bid request. Only a stream that failed to connect with time left counts here; one that ran out the deadline or broke after connecting shows on the status series alone. The fallback's own result is recorded under `endpoint="get_execution_payload_bid"`. | | `cb_pbs_beacon_node_status_code_total` | Counter | `http_status_code`, `endpoint` | HTTP status code returned to the beacon node. Tracks what status codes the PBS returns for beacon node-facing requests. Endpoint values: `get_header`, `register_validator`, `submit_blinded_block`, `status`, `reload`, and (unreleased, from v0.12.0-rc1) `get_execution_payload_bid`, `submit_builder_preferences`, `submit_signed_beacon_block`. Error status codes (`502` for `NoResponse`/`NoPayload`, `500` for `Internal`) are set via `PbsClientError`. The handlers also record `406` directly when the request's `Accept` header offers no supported encoding, `204` when no bid is available on `get_header`, and `202` for accepted v2 `submit_blinded_block` requests. (unreleased, from v0.12.0-rc1) The ePBS endpoints record `200` for a bid, `204` for none, `202` for accepted preferences and for every signed block it forwards, `400` for a request Commit-Boost refuses, a builder's own `400` or `401`, `406` for a bid request whose `Accept` offers no supported encoding, `415` for an unsupported `Content-Type`, and `500` when no builder accepts the preferences. | | `cb_pbs_pbs_submit_block_v2_unsupported_total` | Counter | `relay_id` | Count of v2 `submit_blinded_block` requests a relay could not serve because it returned 404 on the v2 endpoint. A non-zero value means the relay does not support `submitBlindedBlockV2` and those blocks were not submitted via that relay. The double `pbs` in the wire name comes from the registry prefix plus the metric name `pbs_submit_block_v2_unsupported_total`. | diff --git a/docs/docs/get_started/running/metrics.md b/docs/docs/get_started/running/metrics.md index f21ca3cf..6cb6e45f 100644 --- a/docs/docs/get_started/running/metrics.md +++ b/docs/docs/get_started/running/metrics.md @@ -122,3 +122,5 @@ For a relay with `get_header = "stream"` (see [Configuration > Bid streaming](.. | Is the handshake slow? | `cb_pbs_relay_stream_connect_latency` | | Is it actually streaming? | `cb_pbs_relay_stream_updates`: bid updates received per window. A healthy relay sends several; windows that carry at most one update mean the stream connects but does not stream | | Are the frames usable? | `cb_pbs_relay_stream_invalid_frames_total`: frames that could not be parsed as a bid, absent while zero | + +The [ePBS bid stream](../epbs.md#bid-streaming) (unreleased, from v0.12.0-rc1) records its outcome and time-to-first-bid on the same two general series under `endpoint="get_execution_payload_bid_stream"`, and its HTTP fallback under `endpoint="get_execution_payload_bid"`. Its own series, labelled by `relay_id`, are `cb_pbs_relay_epbs_stream_connect_latency`, `cb_pbs_relay_epbs_stream_updates`, `cb_pbs_relay_epbs_stream_invalid_frames_total` and `cb_pbs_relay_epbs_stream_fallback_total`, the counterparts of the four above. It decodes each bid as it arrives and does not validate it, so `cb_pbs_relay_epbs_stream_updates` counts bids that decode and `cb_pbs_relay_epbs_stream_invalid_frames_total` also counts those that do not. A stream that ends with only undecodable bids is no bid (`204`); if its connection drops instead, it records `556`. diff --git a/tests/Cargo.toml b/tests/Cargo.toml index 3e6dfdb4..18ec943e 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -7,6 +7,7 @@ version.workspace = true [dependencies] alloy.workspace = true axum.workspace = true +base64.workspace = true cb-common.workspace = true cb-pbs.workspace = true cb-signer.workspace = true diff --git a/tests/src/mock_relay.rs b/tests/src/mock_relay.rs index e9785d2e..bff94176 100644 --- a/tests/src/mock_relay.rs +++ b/tests/src/mock_relay.rs @@ -337,6 +337,10 @@ impl MockRelayState { } } + pub fn with_trustless_bid_gwei(self, trustless_bid_gwei: u64) -> Self { + Self { trustless_bid_gwei, ..self } + } + /// Hold every bid request `delay_ms` before answering, so a caller with a /// shorter `X-Timeout-Ms` times out. pub fn with_bid_delay_ms(self, delay_ms: u64) -> Self { @@ -403,6 +407,32 @@ pub fn mock_signed_builder_bid( SignedBuilderBid { message, signature } } +/// CB does not verify bid signatures (the beacon node does), so the mock +/// serves an unsigned bid +pub fn mock_execution_payload_bid( + slot: u64, + parent_hash: B256, + parent_root: B256, + value: u64, + execution_payment: u64, +) -> SignedExecutionPayloadBid { + let mut block_hash = B256::ZERO; + block_hash.0[0] = 1; + + let message = ExecutionPayloadBid { + parent_block_hash: parent_hash.into(), + parent_block_root: parent_root, + block_hash: block_hash.into(), + gas_limit: 30_000_000, + builder_index: 42, + slot: Slot::new(slot), + value, + execution_payment, + ..Default::default() + }; + SignedExecutionPayloadBid { message, signature: BlsSignature::empty() } +} + async fn handle_get_execution_payload_bid( State(state): State>, Path((slot, parent_hash, parent_root, _pubkey)): Path<(u64, B256, B256, BlsPublicKey)>, @@ -463,23 +493,8 @@ async fn handle_get_execution_payload_bid( tokio::time::sleep(Duration::from_millis(delay_ms)).await; } - let mut block_hash = B256::ZERO; - block_hash.0[0] = 1; - - let message = ExecutionPayloadBid { - parent_block_hash: parent_hash.into(), - parent_block_root: parent_root, - block_hash: block_hash.into(), - gas_limit: 30_000_000, - builder_index: 42, - slot: Slot::new(slot), - value: state.trustless_bid_gwei, - ..Default::default() - }; - - // CB does not verify bid signatures (the beacon node does), so the mock - // serves an unsigned bid - let data = SignedExecutionPayloadBid { message, signature: BlsSignature::empty() }; + let data = + mock_execution_payload_bid(slot, parent_hash, parent_root, state.trustless_bid_gwei, 0); // Negotiate the RESPONSE encoding from the forwarded Accept, mirroring // handle_get_header: honor supported_content_types + the caller's Accept. diff --git a/tests/src/mock_ws_relay.rs b/tests/src/mock_ws_relay.rs index 2ef44dc0..c45f9aff 100644 --- a/tests/src/mock_ws_relay.rs +++ b/tests/src/mock_ws_relay.rs @@ -1,4 +1,5 @@ use std::{ + net::SocketAddr, str::FromStr, sync::{ Arc, Mutex, @@ -10,10 +11,12 @@ use std::{ use alloy::primitives::{B256, U256}; use cb_common::{ pbs::{ - BuilderApiVersion, GET_HEADER_STREAM_PATH, HEADER_START_TIME_UNIX_MS, HEADER_TIMEOUT_MS, - HEADER_VERSION_KEY, + BuilderApiVersion, GET_EXECUTION_PAYLOAD_BID_STREAM_PATH, GET_HEADER_STREAM_PATH, + HEADER_REQUEST_AUTH, HEADER_START_TIME_UNIX_MS, HEADER_TIMEOUT_MS, HEADER_VERSION_KEY, + RelayClient, }, types::{BlsSecretKey, Chain}, + wire::CONSENSUS_VERSION_HEADER, }; use futures::SinkExt; use ssz::Encode; @@ -22,27 +25,39 @@ use tokio_tungstenite::{ accept_hdr_async, tungstenite::{ Message, - handshake::server::{Request, Response}, + handshake::server::{ErrorResponse, Request, Response}, + http::StatusCode, }, }; use tracing::debug; -use crate::{mock_relay::mock_signed_builder_bid, utils::HEADER_API_KEY}; +use crate::{ + mock_relay::{ + MockRelayState, mock_execution_payload_bid, mock_signed_builder_bid, + start_mock_relay_service_with_listener, + }, + utils::{HEADER_API_KEY, generate_mock_stream_relay, get_free_listener}, +}; const MSG_BID: u8 = 0x01; const FORK_FULU: u8 = 6; +const FORK_GLOAS: u8 = 7; /// What PBS sent in the handshake, captured for assertions. #[derive(Debug, Clone)] pub struct StreamRequest { pub slot: u64, pub parent_hash: B256, + /// Only on the ePBS bid stream + pub parent_root: Option, pub validator_pubkey: String, pub timeout_ms: Option, pub start_time_ms: Option, pub user_agent: Option, pub cb_version: Option, pub api_key: Option, + pub consensus_version: Option, + pub request_auth: Option, } pub struct MockWsRelayState { @@ -51,6 +66,12 @@ pub struct MockWsRelayState { /// One frame pushed per value, in order. The last one is what PBS must /// return. bid_values: Vec, + /// One frame per `(value, execution_payment)` in gwei, on the ePBS bid + /// stream + epbs_bids: Vec<(u64, u64)>, + /// Answer the ePBS stream handshake with 404, as a relay that streams only + /// get_header + no_epbs_stream: bool, /// Pause between updates update_interval: Duration, /// Keep the connection open after the last update, so PBS returns on its @@ -59,6 +80,9 @@ pub struct MockWsRelayState { /// Precede each bid with frames PBS can't parse, which it must skip rather /// than treat as the end of the stream unknown_frames: bool, + /// Cut this many of the last ePBS bids short, so they do not decode + invalid_last_bids: usize, + handshake_attempts: AtomicU64, received_connections: AtomicU64, last_request: Mutex>, } @@ -69,9 +93,13 @@ impl MockWsRelayState { chain, signer, bid_values: vec![U256::from(10)], + epbs_bids: vec![(10, 0)], + no_epbs_stream: false, update_interval: Duration::ZERO, hold_open: false, unknown_frames: false, + invalid_last_bids: 0, + handshake_attempts: AtomicU64::new(0), received_connections: AtomicU64::new(0), last_request: Mutex::new(None), } @@ -81,6 +109,14 @@ impl MockWsRelayState { Self { bid_values, ..self } } + pub fn with_epbs_bids(self, epbs_bids: Vec<(u64, u64)>) -> Self { + Self { epbs_bids, ..self } + } + + pub fn without_epbs_stream(self) -> Self { + Self { no_epbs_stream: true, ..self } + } + pub fn with_update_interval(self, update_interval: Duration) -> Self { Self { update_interval, ..self } } @@ -93,6 +129,16 @@ impl MockWsRelayState { Self { unknown_frames: true, ..self } } + pub fn with_invalid_last_bids(self, invalid_last_bids: usize) -> Self { + Self { invalid_last_bids, ..self } + } + + /// Connections that reached the stream, counted before the handshake, so + /// one that never completes still shows + pub fn handshake_attempts(&self) -> u64 { + self.handshake_attempts.load(Ordering::Relaxed) + } + pub fn received_connections(&self) -> u64 { self.received_connections.load(Ordering::Relaxed) } @@ -120,18 +166,26 @@ pub async fn start_mock_ws_relay_service( // The handshake callback's Err type is fixed by tungstenite #[allow(clippy::result_large_err)] async fn serve_stream(state: Arc, stream: TcpStream) -> eyre::Result<()> { + state.handshake_attempts.fetch_add(1, Ordering::Relaxed); let mut request = None; let mut ws = accept_hdr_async(stream, |req: &Request, res: Response| { request = parse_request(req); + if state.no_epbs_stream && request.as_ref().is_some_and(|req| req.parent_root.is_some()) { + let mut not_found = ErrorResponse::new(None); + *not_found.status_mut() = StatusCode::NOT_FOUND; + return Err(not_found); + } Ok(res) }) .await?; - let request = request.ok_or_else(|| eyre::eyre!("malformed get_header stream request"))?; + let request = request.ok_or_else(|| eyre::eyre!("malformed stream request"))?; state.received_connections.fetch_add(1, Ordering::Relaxed); *state.last_request.lock().unwrap() = Some(request.clone()); - for value in &state.bid_values { + let n_bids = + if request.parent_root.is_some() { state.epbs_bids.len() } else { state.bid_values.len() }; + for i in 0..n_bids { if state.unknown_frames { // Unknown message type, unknown fork, truncated prefix for frame in [vec![0x7f, FORK_FULU, 1], vec![MSG_BID, 0xff, 1], vec![MSG_BID]] { @@ -139,16 +193,8 @@ async fn serve_stream(state: Arc, stream: TcpStream) -> eyre:: } } - let bid = mock_signed_builder_bid( - state.chain, - &state.signer, - request.slot, - request.parent_hash, - *value, - ); - - let mut frame = vec![MSG_BID, FORK_FULU]; - frame.extend_from_slice(&bid.as_ssz_bytes()); + let is_invalid = i + state.invalid_last_bids >= n_bids; + let frame = bid_frame(&state, &request, i, is_invalid); ws.send(Message::Binary(frame.into())).await?; if !state.update_interval.is_zero() { @@ -166,27 +212,131 @@ async fn serve_stream(state: Arc, stream: TcpStream) -> eyre:: Ok(()) } +/// Frame `i` of the stream `request` asked for +fn bid_frame( + state: &MockWsRelayState, + request: &StreamRequest, + i: usize, + invalid: bool, +) -> Vec { + let (fork, bid) = match request.parent_root { + Some(parent_root) => { + let (value, execution_payment) = state.epbs_bids[i]; + let bid = mock_execution_payload_bid( + request.slot, + request.parent_hash, + parent_root, + value, + execution_payment, + ); + let mut bid = bid.as_ssz_bytes(); + if invalid { + bid.truncate(10); + } + (FORK_GLOAS, bid) + } + None => { + let bid = mock_signed_builder_bid( + state.chain, + &state.signer, + request.slot, + request.parent_hash, + state.bid_values[i], + ); + (FORK_FULU, bid.as_ssz_bytes()) + } + }; + + let mut frame = vec![MSG_BID, fork]; + frame.extend_from_slice(&bid); + frame +} + /// The request is the handshake: the fixed stream path followed by -/// `/{slot}/{parent_hash}/{pubkey}`, plus the same headers the HTTP path sends. +/// `/{slot}/{parent_hash}/{pubkey}` (with `/{parent_root}` before the pubkey on +/// the ePBS bid stream), plus the same headers the HTTP path sends. fn parse_request(req: &Request) -> Option { - let prefix = format!("{}{GET_HEADER_STREAM_PATH}/", BuilderApiVersion::V1.path()); - let mut segments = req.uri().path().strip_prefix(&prefix)?.split('/'); + let path = req.uri().path(); + let v1 = BuilderApiVersion::V1.path(); + let (rest, epbs) = + match path.strip_prefix(&format!("{v1}{GET_EXECUTION_PAYLOAD_BID_STREAM_PATH}/")) { + Some(rest) => (rest, true), + None => (path.strip_prefix(&format!("{v1}{GET_HEADER_STREAM_PATH}/"))?, false), + }; + let mut segments = rest.split('/'); let slot = segments.next()?.parse().ok()?; let parent_hash = B256::from_str(segments.next()?).ok()?; + let parent_root = if epbs { Some(B256::from_str(segments.next()?).ok()?) } else { None }; let validator_pubkey = segments.next()?.to_string(); Some(StreamRequest { slot, parent_hash, + parent_root, validator_pubkey, timeout_ms: header(req, HEADER_TIMEOUT_MS).and_then(|v| v.parse().ok()), start_time_ms: header(req, HEADER_START_TIME_UNIX_MS).and_then(|v| v.parse().ok()), user_agent: header(req, "user-agent"), cb_version: header(req, HEADER_VERSION_KEY), api_key: header(req, HEADER_API_KEY), + consensus_version: header(req, CONSENSUS_VERSION_HEADER), + request_auth: header(req, HEADER_REQUEST_AUTH), }) } fn header(req: &Request, name: &str) -> Option { req.headers().get(name)?.to_str().ok().map(ToOwned::to_owned) } + +/// One relay serving its HTTP API and its bid stream on the same port, as a +/// real relay does. Returns the two mock states and a stream-mode client for +/// the relay. +pub async fn start_mock_dual_relay( + http_state: MockRelayState, + ws_state: MockWsRelayState, +) -> eyre::Result<(Arc, Arc, RelayClient)> { + let http_listener = get_free_listener().await; + let http_addr = http_listener.local_addr()?; + let pubkey = http_state.signer.public_key(); + let http_state = Arc::new(http_state); + tokio::spawn(start_mock_relay_service_with_listener(http_state.clone(), http_listener)); + + let listener = get_free_listener().await; + let port = listener.local_addr()?.port(); + let ws_state = Arc::new(ws_state); + tokio::spawn(start_mock_dual_relay_service(ws_state.clone(), listener, http_addr)); + + Ok((http_state, ws_state, generate_mock_stream_relay(port, pubkey)?)) +} + +/// A connection whose request asks for a websocket upgrade goes to the stream, +/// any other is forwarded byte for byte to the HTTP relay at `http_addr`. +async fn start_mock_dual_relay_service( + ws_state: Arc, + listener: TcpListener, + http_addr: SocketAddr, +) -> eyre::Result<()> { + loop { + let (stream, addr) = listener.accept().await?; + let ws_state = ws_state.clone(); + tokio::spawn(async move { + let res = async { + let mut head = [0u8; 2048]; + let n = stream.peek(&mut head).await?; + let head = String::from_utf8_lossy(&head[..n]).to_ascii_lowercase(); + if head.contains("upgrade: websocket") { + serve_stream(ws_state, stream).await + } else { + let mut stream = stream; + let mut upstream = TcpStream::connect(http_addr).await?; + tokio::io::copy_bidirectional(&mut stream, &mut upstream).await?; + Ok(()) + } + } + .await; + if let Err(err) = res { + debug!(%addr, %err, "mock dual relay connection ended"); + } + }); + } +} diff --git a/tests/tests/pbs_get_execution_payload_bid_ws.rs b/tests/tests/pbs_get_execution_payload_bid_ws.rs new file mode 100644 index 00000000..dcaa2ecd --- /dev/null +++ b/tests/tests/pbs_get_execution_payload_bid_ws.rs @@ -0,0 +1,158 @@ +use std::{sync::Arc, time::Duration}; + +use alloy::primitives::B256; +use base64::{Engine, prelude::BASE64_STANDARD}; +use cb_common::{ + pbs::GetExecutionPayloadBidResponse, signer::random_secret, types::Chain, wire::EncodingType, +}; +use cb_tests::{ + mock_relay::MockRelayState, + mock_validator::MockValidator, + mock_ws_relay::{MockWsRelayState, start_mock_dual_relay}, + utils::{RELAY_API_KEY, TEST_AUTH_DATA, opaque_auth, setup_pbs}, +}; +use eyre::Result; +use reqwest::StatusCode; +use ssz::Encode; +use tokio::time::Instant; + +const TEST_SLOT: u64 = 255; +const PARENT_ROOT: B256 = B256::repeat_byte(1); +const BUDGET_MS: u64 = 1_000; + +/// A relay streaming `stream_bids`, each a `(value, execution_payment)` in gwei +fn relay_states(stream_bids: Vec<(u64, u64)>) -> (MockRelayState, MockWsRelayState) { + let signer = random_secret(); + let http_state = MockRelayState::new(Chain::Hoodi, signer.clone()); + (http_state, MockWsRelayState::new(Chain::Hoodi, signer).with_epbs_bids(stream_bids)) +} + +/// PBS in front of one stream-mode relay serving both states on one port +async fn start_relay( + (http_state, ws_state): (MockRelayState, MockWsRelayState), +) -> Result<(MockValidator, Arc, Arc)> { + let (http_state, ws_state, relay) = start_mock_dual_relay(http_state, ws_state).await?; + let validator = setup_pbs(Chain::Hoodi, vec![relay], |_| {}).await?; + Ok((validator, http_state, ws_state)) +} + +async fn get_bid( + validator: &MockValidator, +) -> Result<(StatusCode, Option)> { + let res = validator + .do_get_execution_payload_bid_with_timeout( + TEST_SLOT, + B256::ZERO, + PARENT_ROOT, + None, + Some(&opaque_auth(TEST_AUTH_DATA, TEST_SLOT)), + vec![EncodingType::Json], + BUDGET_MS, + ) + .await?; + let code = res.status(); + if code != StatusCode::OK { + return Ok((code, None)); + } + Ok((code, Some(serde_json::from_slice(&res.bytes().await?)?))) +} + +fn assert_bid(res: Option, value: u64, execution_payment: u64) { + let res = res.expect("a bid"); + let bid = &res.data.message; + assert_eq!((bid.value, bid.execution_payment), (value, execution_payment)); +} + +/// A relay that streams get_header but not the ePBS bid answers the handshake +/// with 404, and PBS falls back to HTTP: its bid is served, and the builder's +/// 401 reaches the beacon node. +#[tokio::test] +async fn test_get_execution_payload_bid_ws_no_stream_route_falls_back_to_http() -> Result<()> { + for rejection in [None, Some(StatusCode::UNAUTHORIZED)] { + let (http_state, ws_state) = relay_states(vec![]); + let (validator, http_state, ws_state) = + start_relay((http_state.with_trustless_bid_gwei(42), ws_state.without_epbs_stream())) + .await?; + if let Some(status) = rejection { + http_state.set_response_override(status); + } + + let (code, res) = get_bid(&validator).await?; + assert_eq!(code, rejection.unwrap_or(StatusCode::OK)); + if rejection.is_none() { + assert_bid(res, 42, 0); + } + assert_eq!(http_state.received_execution_payload_bid(), 1); + assert_eq!(ws_state.handshake_attempts(), 1); + } + Ok(()) +} + +/// A relay whose stream serves a bid gets no HTTP request, and its latest bid +/// is served, not its highest. +#[tokio::test] +async fn test_get_execution_payload_bid_ws_serves_latest_bid_without_http() -> Result<()> { + let (validator, http_state, ws_state) = + start_relay(relay_states(vec![(70, 0), (50, 0)])).await?; + + let (code, res) = get_bid(&validator).await?; + assert_eq!(code, StatusCode::OK); + assert_bid(res, 50, 0); + assert_eq!(http_state.received_execution_payload_bid(), 0); + assert_eq!(ws_state.received_connections(), 1); + Ok(()) +} + +/// A later streamed bid that fails to decode does not displace the valid bid +/// (50) before it. +#[tokio::test] +async fn test_get_execution_payload_bid_ws_skips_invalid_bid() -> Result<()> { + let (http_state, ws_state) = relay_states(vec![(50, 0), (90, 0)]); + let (validator, ..) = start_relay((http_state, ws_state.with_invalid_last_bids(1))).await?; + + let (code, res) = get_bid(&validator).await?; + assert_eq!(code, StatusCode::OK); + assert_bid(res, 50, 0); + Ok(()) +} + +/// The handshake carries the auth as padded standard base64 of its SSZ, the +/// fork and a timeout that leaves the buffer; the bid lands within the budget. +#[tokio::test] +async fn test_get_execution_payload_bid_ws_handshake_and_deadline() -> Result<()> { + const BUFFER_MS: u64 = 200; + let (http_state, ws_state) = relay_states(vec![(20, 0)]); + let (_, ws_state, relay) = start_mock_dual_relay(http_state, ws_state.hold_open()).await?; + let validator = setup_pbs(Chain::Hoodi, vec![relay], |pbs_config| { + pbs_config.proposer_deadline_buffer_ms = BUFFER_MS + }) + .await?; + + let start = Instant::now(); + let sent_at = cb_common::utils::utcnow_ms(); + let (code, _) = get_bid(&validator).await?; + assert_eq!(code, StatusCode::OK); + assert!(start.elapsed() < Duration::from_millis(BUDGET_MS), "{:?}", start.elapsed()); + + let request = ws_state.last_request().expect("a handshake"); + // 119 bytes of SSZ need padding, and slot 255 puts a '/' in the base64, + // which the URL-safe alphabet would write as '_' + let expected_auth = + BASE64_STANDARD.encode(opaque_auth(TEST_AUTH_DATA, TEST_SLOT).as_ssz_bytes()); + assert!(expected_auth.ends_with('=') && expected_auth.contains('/')); + assert_eq!(request.request_auth, Some(expected_auth)); + + assert_eq!(request.slot, TEST_SLOT); + assert_eq!(request.parent_root, Some(PARENT_ROOT)); + assert_eq!(request.consensus_version.as_deref(), Some("gloas")); + let relay_timeout_ms = request.timeout_ms.expect("missing timeout header"); + assert!( + (BUDGET_MS - BUFFER_MS - 100..=BUDGET_MS - BUFFER_MS).contains(&relay_timeout_ms), + "timeout header {relay_timeout_ms}ms" + ); + let start_time_ms = request.start_time_ms.expect("missing start time header"); + assert!((sent_at..sent_at + BUDGET_MS).contains(&start_time_ms), "{start_time_ms}"); + assert_eq!(request.api_key.as_deref(), Some(RELAY_API_KEY)); + assert!(request.user_agent.is_some_and(|ua| ua.contains("commit-boost"))); + Ok(()) +}