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..f4a177bf 100644 --- a/config.example.toml +++ b/config.example.toml @@ -100,13 +100,17 @@ 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. +# How to fetch bids 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):///eth/v1/builder/header_stream/{slot}/{parent_hash}/{pubkey}. +# ws(s):///eth/v1/builder/header_stream/{slot}/{parent_hash}/{pubkey}. +# (unreleased, from v0.12.0-rc1) The relay's normal HTTP request runs alongside, +# timing games included, and the better bid wins, so a failed stream still +# leaves the HTTP bid. # 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/config/pbs.rs b/crates/common/src/config/pbs.rs index 294c5b53..00df7f6e 100644 --- a/crates/common/src/config/pbs.rs +++ b/crates/common/src/config/pbs.rs @@ -49,6 +49,15 @@ pub enum GetHeaderTransport { Stream, } +impl GetHeaderTransport { + pub fn as_str(&self) -> &'static str { + match self { + Self::Http => "http", + Self::Stream => "stream", + } + } +} + /// A custom relay header value: a literal, or a secret read from a file or an /// environment variable when the relay client is built (at startup and on every /// reload), so an API key never has to sit in plaintext in the config file. @@ -131,7 +140,7 @@ pub struct RelayConfig { pub headers: Option>, /// Optional GET parameters to add to each request pub get_params: Option>, - /// How to fetch headers from this relay + /// How to fetch bids from this relay #[serde(default)] pub get_header: GetHeaderTransport, /// Whether to enable timing games 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 b877d858..4a708b14 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, }; @@ -63,6 +63,15 @@ pub enum GetHeaderRequest { Stream(Url), } +impl GetHeaderRequest { + pub fn transport(&self) -> GetHeaderTransport { + match self { + Self::Http(_) => GetHeaderTransport::Http, + Self::Stream(_) => GetHeaderTransport::Stream, + } + } +} + fn stream_url(entry: &Url) -> eyre::Result { let scheme = match entry.scheme() { "http" | "ws" => "ws", @@ -87,9 +96,10 @@ pub struct RelayClient { pub id: Arc, /// HTTP client to send requests pub client: reqwest::Client, - /// Base url of the get_header stream, `Some` only when the relay streams. + /// Base url of the get_header stream, whose origin the ePBS bid stream + /// shares. `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, @@ -195,24 +205,17 @@ impl RelayClient { ) } - pub fn get_header_request( + /// The get_header stream, `None` when the relay does not stream + pub fn get_header_stream_url( &self, slot: u64, parent_hash: &B256, validator_pubkey: &BlsPublicKey, - ) -> Result { - Ok(match &self.stream_url { - None => { - GetHeaderRequest::Http(self.get_header_url(slot, parent_hash, validator_pubkey)?) - } - Some(base) => { - let mut url = base.clone(); - url.set_path(&format!("{}/{slot}/{parent_hash}/{validator_pubkey}", base.path())); - - self.append_get_params(&mut url); - GetHeaderRequest::Stream(url) - } - }) + ) -> Option { + let mut url = self.stream_url.clone()?; + url.set_path(&format!("{}/{slot}/{parent_hash}/{validator_pubkey}", url.path())); + self.append_get_params(&mut url); + Some(url) } pub fn get_status_url(&self) -> Result { @@ -244,6 +247,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, @@ -281,9 +302,7 @@ mod tests { use alloy::primitives::B256; - use super::{ - GetHeaderRequest, RelayClient, RelayEntry, decode_auth_data_url, value_fingerprint, - }; + use super::{RelayClient, RelayEntry, decode_auth_data_url, value_fingerprint}; use crate::{ config::{GetHeaderTransport, RelayConfig, test_env::RELAY_URL}, utils::bls_pubkey_from_hex_unchecked, @@ -391,9 +410,10 @@ mod tests { } #[test] - fn test_get_header_request() { + fn test_stream_urls() { let slot = 0; let parent_hash = B256::ZERO; + let parent_root = B256::repeat_byte(1); let validator_pubkey = bls_pubkey_from_hex_unchecked( "0xac6e77dfe25ecd6110b8e780608cce0dab71fdd5ebea22a16c0205200f2f8e2e3ad3b71d3499c54ad14d6c21b41a37ae", ); @@ -403,17 +423,18 @@ mod tests { }"#; let base_config = serde_json::from_str::(relay_config).unwrap(); - // Default transport: plain HTTP endpoint + // Default transport: no stream let relay = RelayClient::new(base_config.clone()).unwrap(); - let GetHeaderRequest::Http(url) = - relay.get_header_request(slot, &parent_hash, &validator_pubkey).unwrap() - else { - panic!("expected http request"); - }; - assert_eq!( - url, - relay.get_header_url(slot, &parent_hash, &validator_pubkey).unwrap(), - "http dispatch must match the plain url builder" + assert!(relay.get_header_stream_url(slot, &parent_hash, &validator_pubkey).is_none()); + assert!( + relay + .get_execution_payload_bid_stream_url( + slot, + &parent_hash, + &parent_root, + &validator_pubkey + ) + .is_none() ); // Streaming: the relay url over ws, at the fixed stream path, with the @@ -421,11 +442,7 @@ mod tests { let mut config = base_config.clone(); config.get_header = GetHeaderTransport::Stream; let relay = RelayClient::new(config).unwrap(); - let GetHeaderRequest::Stream(url) = - relay.get_header_request(slot, &parent_hash, &validator_pubkey).unwrap() - else { - panic!("expected stream request"); - }; + let url = relay.get_header_stream_url(slot, &parent_hash, &validator_pubkey).unwrap(); assert_eq!( url.to_string(), format!( @@ -440,17 +457,27 @@ mod tests { config.get_header = GetHeaderTransport::Stream; config.get_params = Some(HashMap::from([("token".to_string(), "abc".to_string())])); let relay = RelayClient::new(config).unwrap(); - let GetHeaderRequest::Stream(url) = - relay.get_header_request(slot, &parent_hash, &validator_pubkey).unwrap() - else { - panic!("expected stream request"); - }; + let url = relay.get_header_stream_url(slot, &parent_hash, &validator_pubkey).unwrap(); assert_eq!( url.to_string(), format!( "wss://abc.xyz:4444/eth/v1/builder/header_stream/{slot}/{parent_hash}/{validator_pubkey}?token=abc" ) ); + let url = relay + .get_execution_payload_bid_stream_url( + slot, + &parent_hash, + &parent_root, + &validator_pubkey, + ) + .unwrap(); + assert_eq!( + url.to_string(), + format!( + "wss://abc.xyz:4444/eth/v1/builder/execution_payload_bid_stream/{slot}/{parent_hash}/{parent_root}/{validator_pubkey}?token=abc" + ) + ); // A relay url we can't stream over is rejected at construction let mut config = base_config; 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/bid_stream.rs b/crates/pbs/src/bid_stream.rs new file mode 100644 index 00000000..6dd63002 --- /dev/null +++ b/crates/pbs/src/bid_stream.rs @@ -0,0 +1,452 @@ +//! The websocket bid stream of a relay configured with `get_header = "stream"`. +//! One connection per request, dropped at the deadline. The handshake carries +//! the request, and the relay replies with one binary frame per bid update: +//! +//! ```text +//! u8 message type +//! u8 fork +//! .. SSZ bid +//! ``` + +use std::{ + collections::VecDeque, + io, + sync::{Arc, OnceLock}, + time::Duration, +}; + +use axum::http::{HeaderMap, HeaderValue, Request, Uri, header::USER_AGENT}; +use cb_common::{ + pbs::{ForkName, HEADER_START_TIME_UNIX_MS, HEADER_TIMEOUT_MS, RelayClient, error::PbsError}, + utils::utcnow_ms, +}; +use futures::StreamExt; +use reqwest::StatusCode; +use rustls::{ClientConfig, RootCertStore, crypto::aws_lc_rs}; +use tokio::{ + net::{TcpStream, lookup_host}, + time::{Instant, sleep_until, timeout_at}, +}; +use tokio_tungstenite::{ + Connector, MaybeTlsStream, WebSocketStream, client_async_tls_with_config, + tungstenite::{ + Bytes, Error as WsError, Message, client::IntoClientRequest, error::UrlError, + protocol::WebSocketConfig, + }, +}; +use tracing::{debug, warn}; +use url::Url; + +use crate::{ + constants::{MAX_SIZE_GET_HEADER_RESPONSE, TIMEOUT_ERROR_STATUS, TRANSPORT_ERROR_STATUS}, + metrics::{ + RELAY_LATENCY, RELAY_STREAM_CONNECT_LATENCY, RELAY_STREAM_FALLBACK, + RELAY_STREAM_INVALID_FRAMES, RELAY_STREAM_UPDATES, + }, +}; + +/// Frame prefix: message type + fork. +const FRAME_PREFIX_LEN: usize = 2; + +const MSG_BID: u8 = 0x01; + +/// A PBS stream holds only its newest frames, which bounds its memory and the +/// validation it leaves until the deadline +pub(crate) const MAX_HELD_FRAMES: usize = 8; + +/// One bid update, its bid still SSZ-encoded +pub(crate) struct Frame { + pub(crate) fork: ForkName, + pub(crate) bid: Bytes, +} + +/// What a stream held when its window closed +pub(crate) struct Held { + /// The newest frames the parse closure accepted, up to the caller's cap, + /// oldest first + pub(crate) frames: VecDeque, + /// Frames the parse closure accepted + pub(crate) updates: usize, + pub(crate) connect_latency: Duration, + pub(crate) first_frame_latency: Option, + pub(crate) invalid_frames: usize, +} + +/// The handshake for `url`, with the caller's user agent and the relay's +/// headers +pub(crate) fn handshake_request( + url: &Url, + relay: &RelayClient, + send_headers: &HeaderMap, +) -> 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()); + } + + Ok(request) +} + +/// Holds the newest `max_held` frames `parse` accepts until `deadline` or the +/// end of the stream. An error carries the status the failure records under. +pub(crate) async fn read_bid_stream( + request: Request<()>, + deadline: Instant, + relay: &RelayClient, + endpoint: &str, + max_held: usize, + mut parse: impl FnMut(Frame) -> Result, +) -> Result, (StatusCode, PbsError)> { + let start_request = Instant::now(); + let mut stream = match timeout_at(deadline, connect(request, deadline)).await { + Ok(Ok(Some(stream))) => stream, + // A relay with no bid can answer the handshake as it answers get_header + Ok(Err(WsError::Http(res))) if res.status() == StatusCode::NO_CONTENT => { + return Ok(Held { + frames: VecDeque::new(), + updates: 0, + connect_latency: start_request.elapsed(), + first_frame_latency: None, + invalid_frames: 0, + }); + } + Ok(Err(err)) => return Err(connect_failed(&err)), + Ok(Ok(None)) | Err(_) => return Err((TIMEOUT_ERROR_STATUS, PbsError::WebSocketTimeout)), + }; + let connect_latency = start_request.elapsed(); + RELAY_STREAM_CONNECT_LATENCY + .with_label_values(&[endpoint, relay.id.as_str()]) + .observe(connect_latency.as_secs_f64()); + debug!(relay_id = relay.id.as_ref(), endpoint, ?connect_latency, "ws connected"); + + let timer = sleep_until(deadline); + tokio::pin!(timer); + + let mut frames = VecDeque::with_capacity(max_held); + let mut updates = 0usize; + let mut first_frame_latency = None; + let mut invalid_frames = 0usize; + let mut stream_error = None; + + loop { + let message = tokio::select! { + biased; + _ = &mut timer => break, + message = stream.next() => message, + }; + + let payload = match message { + Some(Ok(Message::Binary(payload))) => payload, + Some(Ok(Message::Close(_))) | None => break, + Some(Ok(_)) => continue, + Some(Err(err)) => { + debug!(relay_id = relay.id.as_ref(), %err, "ws stream error"); + stream_error = Some(PbsError::WebSocket(format!("stream error: {err}"))); + break; + } + }; + + match parse_frame(payload).and_then(&mut parse) { + Ok(frame) => { + first_frame_latency.get_or_insert_with(|| start_request.elapsed()); + updates += 1; + if frames.len() == max_held { + frames.pop_front(); + } + frames.push_back(frame); + } + Err(err) => { + invalid_frames += 1; + if invalid_frames == 1 { + warn!(relay_id = relay.id.as_ref(), %err, "invalid ws frame, skipping"); + } + } + } + } + + drop(stream); + + RELAY_STREAM_UPDATES.with_label_values(&[endpoint, relay.id.as_str()]).observe(updates as f64); + if invalid_frames > 0 { + RELAY_STREAM_INVALID_FRAMES + .with_label_values(&[endpoint, relay.id.as_str()]) + .inc_by(invalid_frames as u64); + } + + if let Some(latency) = first_frame_latency { + RELAY_LATENCY.with_label_values(&[endpoint, &relay.id]).observe(latency.as_secs_f64()); + } else if let Some(err) = stream_error { + return Err((TRANSPORT_ERROR_STATUS, err)); + } + + Ok(Held { frames, updates, connect_latency, first_frame_latency, invalid_frames }) +} + +/// Opens TCP first and times it, so `X-Timeout-Ms` ends the relay's stream one +/// downlink trip before `deadline`. None when no time would be left. +async fn connect( + mut request: Request<()>, + deadline: Instant, +) -> Result>>, WsError> { + let (socket, rtt) = connect_tcp(request.uri()).await?; + + let Some(timeout_ms) = + relay_timeout_ms(deadline.saturating_duration_since(Instant::now()), rtt) + else { + return Ok(None); + }; + let headers = request.headers_mut(); + headers.insert(HEADER_START_TIME_UNIX_MS, HeaderValue::from(utcnow_ms())); + headers.insert(HEADER_TIMEOUT_MS, HeaderValue::from(timeout_ms)); + + let config = WebSocketConfig::default() + .max_message_size(Some(MAX_SIZE_GET_HEADER_RESPONSE)) + .max_frame_size(Some(MAX_SIZE_GET_HEADER_RESPONSE)); + let connector = Connector::Rustls(tls_config().clone()); + let (stream, _) = + client_async_tls_with_config(request, socket, Some(config), Some(connector)).await?; + + Ok(Some(stream)) +} + +/// The TCP handshake's duration is one round trip, free of clock skew. Name +/// resolution is left out of it. +async fn connect_tcp(uri: &Uri) -> Result<(TcpStream, Duration), WsError> { + let host = uri.host().ok_or(WsError::Url(UrlError::NoHostName))?; + let host = host.strip_prefix('[').and_then(|host| host.strip_suffix(']')).unwrap_or(host); + let port = uri + .port_u16() + .or(match uri.scheme_str() { + Some("wss") => Some(443), + Some("ws") => Some(80), + _ => None, + }) + .ok_or(WsError::Url(UrlError::UnsupportedUrlScheme))?; + + let mut last_err = None; + for addr in lookup_host((host, port)).await? { + let start = Instant::now(); + match TcpStream::connect(addr).await { + Ok(socket) => { + let rtt = start.elapsed(); + socket.set_nodelay(true)?; + return Ok((socket, rtt)); + } + Err(err) => last_err = Some(err), + } + } + + Err(WsError::Io(last_err.unwrap_or_else(|| { + io::Error::new(io::ErrorKind::InvalidInput, "could not resolve to any address") + }))) +} + +/// The time left less half the round trip, the estimate of one downlink trip +fn relay_timeout_ms(time_left: Duration, rtt: Duration) -> Option { + let timeout_ms = time_left.saturating_sub(rtt / 2).as_millis() as u64; + (timeout_ms > 0).then_some(timeout_ms) +} + +pub(crate) fn record_stream_failure(endpoint: &str, relay_id: &str, err: &PbsError) { + warn!(%err, relay_id, endpoint, "stream failed"); + RELAY_STREAM_FALLBACK.with_label_values(&[endpoint, relay_id]).inc(); +} + +/// 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) -> (StatusCode, PbsError) { + let WsError::Http(res) = err else { + return (TRANSPORT_ERROR_STATUS, 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, PbsError::WebSocketConnect(msg)) +} + +fn fork_from_wire(byte: u8) -> Option { + // TODO @nina: I don't see a point of extending a u8 for supporting older forks + // we could rotate these instead, i.e. 0 becomes Heze, etc + 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, + }) +} + +fn parse_frame(payload: Bytes) -> Result { + 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(Frame { fork, bid: payload.slice(FRAME_PREFIX_LEN..) }) +} + +/// One TLS config for every 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_relay_timeout_leaves_the_downlink_trip() { + let ms = Duration::from_millis; + assert_eq!(relay_timeout_ms(ms(1_000), ms(100)), Some(950)); + assert_eq!(relay_timeout_ms(ms(1_000), Duration::ZERO), Some(1_000)); + // Nothing left: no frame could land before the deadline + assert_eq!(relay_timeout_ms(ms(50), ms(100)), None); + assert_eq!(relay_timeout_ms(ms(40), ms(100)), None); + } + + // The host of an IPv6 url keeps its brackets, which name resolution rejects + #[tokio::test] + async fn test_connect_tcp_to_an_ipv6_literal() { + let Ok(listener) = tokio::net::TcpListener::bind("[::1]:0").await else { + return; // no IPv6 loopback on this host + }; + let port = listener.local_addr().unwrap().port(); + let uri: Uri = format!("ws://[::1]:{port}/").parse().unwrap(); + assert!(connect_tcp(&uri).await.is_ok()); + } + + #[test] + fn test_parse_frame() { + assert!(matches!( + parse_frame(bid_frame(6, &[1, 2, 3])), + Ok(Frame { fork: 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(Frame { fork: 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, err) = connect_failed(&WsError::Http(Box::new(rejected))); + assert_eq!(status, StatusCode::UNAUTHORIZED); + let PbsError::WebSocketConnect(msg) = err 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, err) = connect_failed(&WsError::ConnectionClosed); + assert_eq!(status, TRANSPORT_ERROR_STATUS); + assert!(matches!(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, 299] { + let answered = axum::http::Response::builder().status(code).body(None).unwrap(); + let (status, err) = connect_failed(&WsError::Http(Box::new(answered))); + assert_eq!( + status, TRANSPORT_ERROR_STATUS, + "handshake answered {code} counted as a served stream" + ); + let PbsError::WebSocketConnect(msg) = err 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/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 d5a75a00..b93938be 100644 --- a/crates/pbs/src/dial.rs +++ b/crates/pbs/src/dial.rs @@ -174,6 +174,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, @@ -292,6 +293,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/lib.rs b/crates/pbs/src/lib.rs index 9c04224a..19acfc2f 100644 --- a/crates/pbs/src/lib.rs +++ b/crates/pbs/src/lib.rs @@ -1,4 +1,5 @@ mod api; +mod bid_stream; mod constants; mod dial; mod error; diff --git a/crates/pbs/src/metrics.rs b/crates/pbs/src/metrics.rs index 51846f94..9745d53e 100644 --- a/crates/pbs/src/metrics.rs +++ b/crates/pbs/src/metrics.rs @@ -52,46 +52,46 @@ lazy_static! { .unwrap(); - // THE WEBSOCKET BID STREAM + // THE WEBSOCKET BID STREAMS // Outcome and time-to-first-bid ride RELAY_STATUS_CODE / RELAY_LATENCY - // under `get_header_stream` + // under `get_header_stream` and `get_execution_payload_bid_stream` /// Websocket handshake latency by relay pub static ref RELAY_STREAM_CONNECT_LATENCY: HistogramVec = register_histogram_vec_with_registry!( "relay_stream_connect_latency", "Websocket handshake latency by relay", - &["relay_id"], + &["endpoint", "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(); - /// Bid updates received per stream window by relay + /// Stream frames accepted on arrival per window, by relay + // On arrival get_header_stream checks only the frame prefix, while + // get_execution_payload_bid_stream decodes the bid pub static ref RELAY_STREAM_UPDATES: HistogramVec = register_histogram_vec_with_registry!( "relay_stream_updates", - "Bid updates received per stream window by relay", - &["relay_id"], + "Stream frames accepted on arrival per window, by relay", + &["endpoint", "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(); - /// Websocket frames that could not be parsed as a bid, by relay + /// Stream frames rejected as invalid on arrival, by relay pub static ref RELAY_STREAM_INVALID_FRAMES: IntCounterVec = register_int_counter_vec_with_registry!( "relay_stream_invalid_frames_total", - "Websocket frames that could not be parsed as a bid, by relay", - &["relay_id"], + "Stream frames rejected as invalid on arrival, by relay", + &["endpoint", "relay_id"], PBS_METRICS_REGISTRY ) .unwrap(); - /// Stream attempts that fell back to HTTP, by relay - // Only a handshake failure with bid window left retries; one at the - // deadline shows on the status series alone + /// Bid streams that failed while the relay's HTTP request ran alongside pub static ref RELAY_STREAM_FALLBACK: IntCounterVec = register_int_counter_vec_with_registry!( "relay_stream_fallback_total", - "get_header stream attempts that fell back to HTTP, by relay", - &["relay_id"], + "Bid streams that failed while the relay's HTTP request ran alongside, by relay", + &["endpoint", "relay_id"], PBS_METRICS_REGISTRY ) .unwrap(); diff --git a/crates/pbs/src/mev_boost/get_header.rs b/crates/pbs/src/mev_boost/get_header.rs index 670e80b1..19067d52 100644 --- a/crates/pbs/src/mev_boost/get_header.rs +++ b/crates/pbs/src/mev_boost/get_header.rs @@ -1,4 +1,6 @@ use std::{ + collections::HashMap, + iter::once, sync::Arc, time::{Duration, Instant}, }; @@ -10,6 +12,7 @@ use alloy::{ }; use axum::http::{HeaderMap, HeaderValue}; use cb_common::{ + config::GetHeaderTransport, constants::APPLICATION_BUILDER_DOMAIN, pbs::{ EMPTY_TX_ROOT_HASH, ExecutionPayloadHeaderRef, ForkName, ForkVersionDecode, GetHeaderInfo, @@ -25,7 +28,7 @@ use cb_common::{ parse_response_encoding_and_fork, safe_read_http_response, }, }; -use futures::future::join_all; +use futures::{FutureExt, future::join_all}; use parking_lot::RwLock; use reqwest::{ StatusCode, @@ -38,14 +41,12 @@ use url::Url; use super::get_header_ws::get_header_ws; use crate::{ + bid_stream::record_stream_failure, constants::{ - GET_HEADER_ENDPOINT_TAG, MAX_SIZE_GET_HEADER_RESPONSE, TIMEOUT_ERROR_CODE, - TIMEOUT_ERROR_CODE_STR, - }, - metrics::{ - RELAY_HEADER_VALUE, RELAY_LAST_SLOT, RELAY_LATENCY, RELAY_STATUS_CODE, - RELAY_STREAM_FALLBACK, + GET_HEADER_ENDPOINT_TAG, GET_HEADER_STREAM_ENDPOINT_TAG, MAX_SIZE_GET_HEADER_RESPONSE, + TIMEOUT_ERROR_CODE, TIMEOUT_ERROR_CODE_STR, }, + metrics::{RELAY_HEADER_VALUE, RELAY_LAST_SLOT, RELAY_LATENCY, RELAY_STATUS_CODE}, state::{BuilderApiState, PbsState}, utils::check_gas_limit, }; @@ -171,52 +172,77 @@ pub async fn get_header( parent_block, }, }); + // A stream relay races its normal HTTP request; each is a separate candidate + // below + let params = &request_info.params; let mut handles = Vec::with_capacity(relays.len()); for relay in relays.iter() { - handles.push( - get_header_from_relay( - request_info.clone(), - relay.clone(), - ms_into_slot, - max_timeout_ms, - ) - .in_current_span(), - ); + let relay_id = relay.id.as_str(); + let http_url = match relay.get_header_url(params.slot, ¶ms.parent_hash, ¶ms.pubkey) + { + Ok(url) => url, + Err(err) => { + error!(%err, relay_id); + continue; + } + }; + let stream_url = + relay.get_header_stream_url(params.slot, ¶ms.parent_hash, ¶ms.pubkey); + + let requests = + once(GetHeaderRequest::Http(http_url)).chain(stream_url.map(GetHeaderRequest::Stream)); + for request in requests { + let transport = request.transport(); + handles.push( + get_header_from_relay( + request_info.clone(), + relay.clone(), + ms_into_slot, + max_timeout_ms, + request, + ) + .map(move |res| (relay_id, transport, res)) + .in_current_span(), + ); + } } - let results = join_all(handles).await; - let mut relay_bids = Vec::with_capacity(relays.len()); - for (i, res) in results.into_iter().enumerate() { - let relay_id = relays[i].id.as_str(); - + let mut relay_bids = Vec::with_capacity(handles.len()); + let mut best_per_relay: HashMap<&str, U256> = HashMap::new(); + for (relay_id, transport, res) in join_all(handles).await { match res { Ok(Some(res)) => { - RELAY_LAST_SLOT.with_label_values(&[relay_id]).set(slot); - let value_gwei = (res.data.message.value() / U256::from(1_000_000_000)) - .try_into() - .unwrap_or_default(); - RELAY_HEADER_VALUE.with_label_values(&[relay_id]).set(value_gwei); - - relay_bids.push((relay_id, res)) + let best = best_per_relay.entry(relay_id).or_default(); + *best = (*best).max(*res.value()); + relay_bids.push((relay_id, transport, res)) + } + Ok(None) => {} + Err(err) if transport == GetHeaderTransport::Stream => { + record_stream_failure(GET_HEADER_STREAM_ENDPOINT_TAG, relay_id, &err) } - Ok(_) => {} Err(err) if err.is_timeout() => error!(err = "Timed Out", relay_id), Err(err) => error!(%err, relay_id), } } + for (relay_id, value) in best_per_relay { + RELAY_LAST_SLOT.with_label_values(&[relay_id]).set(slot); + let value_gwei = (value / U256::from(1_000_000_000)).try_into().unwrap_or_default(); + RELAY_HEADER_VALUE.with_label_values(&[relay_id]).set(value_gwei); + } - let max_bid = relay_bids.into_iter().max_by_key(|(_, bid)| *bid.value()); + let max_bid = relay_bids.into_iter().max_by_key(|(_, _, bid)| *bid.value()); - if let Some((winning_relay_id, ref bid)) = max_bid { + if let Some((winning_relay_id, transport, ref bid)) = max_bid { info!( relay_id = winning_relay_id, + transport = transport.as_str(), value_eth = format_ether(*bid.value()), block_hash = %bid.block_hash(), "auction winner" ); } - Ok(max_bid.map(|(_, bid)| bid)) + Ok(max_bid.map(|(_, _, bid)| bid)) } /// Fetch the parent block from the RPC URL for extra validation of the header. @@ -249,35 +275,11 @@ async fn get_header_from_relay( relay: RelayClient, ms_into_slot: u64, timeout_left_ms: u64, + request: GetHeaderRequest, ) -> Result, PbsError> { - let params = &request_info.params; - - match relay.get_header_request(params.slot, ¶ms.parent_hash, ¶ms.pubkey)? { + match request { GetHeaderRequest::Stream(url) => { - 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)); - } - - RELAY_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 get_header"); - - let url = relay.get_header_url(params.slot, ¶ms.parent_hash, ¶ms.pubkey)?; - send_timed_get_header( - request_info, - relay, - ms_into_slot + elapsed_ms, - url, - timeout_left_ms, - ) - .await + get_header_ws(&request_info, &relay, url, timeout_left_ms).await } GetHeaderRequest::Http(url) => { send_timed_get_header(request_info, relay, ms_into_slot, url, timeout_left_ms).await @@ -733,19 +735,24 @@ fn extra_validation( #[cfg(test)] mod tests { - use std::{fs, path::Path}; + use std::{fs, net::SocketAddr, path::Path}; use alloy::primitives::{B256, U256}; use cb_common::{ + config::{PbsConfig, PbsModuleConfig, RelayConfig}, pbs::*, signature::sign_builder_message, ssz::get_bid_value_from_signed_builder_bid_ssz, types::{BlsPublicKeyBytes, BlsSecretKey, BlsSignature, Chain}, utils::{TestRandomSeed, timestamp_of_slot_start_sec}, }; + use futures::SinkExt; use ssz::Encode; + use tokio::net::TcpListener; + use tokio_tungstenite::{accept_async, tungstenite::Message}; use super::{validate_header_data, *}; + use crate::metrics::{RELAY_STREAM_FALLBACK, RELAY_STREAM_UPDATES}; #[test] fn test_validate_header() { @@ -881,4 +888,207 @@ mod tests { assert_eq!(*decoded.value(), bid_value); } } + + /// A relay that answers the stream handshake with `status`, and its HTTP + /// get_header with 204, so only the stream can count as failed + async fn start_status_relay(status: StatusCode) -> SocketAddr { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let relay = axum::Router::new().fallback(move |headers: HeaderMap| async move { + if headers.contains_key(reqwest::header::UPGRADE) { + status + } else { + StatusCode::NO_CONTENT + } + }); + tokio::spawn(async move { axum::serve(listener, relay).await }); + addr + } + + /// A relay whose stream sends `frames` copies of a bid for another parent, + /// then closes. Its HTTP get_header fails. + async fn start_invalid_bid_stream_relay(frames: usize) -> SocketAddr { + let json = fs::read("../../tests/data/get_header/fulu.json").unwrap(); + // Message type bid, fork fulu + let mut frame = vec![0x01, 6]; + frame.extend(decode_json_payload(&json).unwrap().data.as_ssz_bytes()); + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + while let Ok((socket, _)) = listener.accept().await { + let frame = frame.clone(); + tokio::spawn(async move { + let Ok(mut stream) = accept_async(socket).await else { return }; + for _ in 0..frames { + let _ = stream.send(Message::Binary(frame.clone().into())).await; + } + let _ = stream.close(None).await; + }); + } + }); + addr + } + + /// get_header for `slot` and `parent_hash` against one stream relay at + /// `addr`. Returns the relay's id, which labels its metrics, and the bid. + async fn get_header_from_stream_relay( + addr: SocketAddr, + slot: u64, + parent_hash: B256, + ) -> (String, Option) { + let pubkey = BlsSecretKey::random().public_key(); + let relay_id = format!("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 mut pbs_config: PbsConfig = serde_json::from_str("{}").unwrap(); + pbs_config.late_in_slot_time_ms = u64::MAX; + // The fixture bids are not signed by the relay's key + pbs_config.skip_sigverify = true; + let state = PbsState::new( + PbsModuleConfig { + chain: Chain::Holesky, + endpoint: addr, + pbs_config: Arc::new(pbs_config), + relays: vec![relay.clone()], + all_relays: vec![relay], + signer_client: None, + registry_muxes: None, + mux_lookup: None, + }, + Default::default(), + ); + let params = GetHeaderParams { slot, parent_hash, pubkey }; + + let bid = get_header(params, HeaderMap::new(), state).await.unwrap(); + (relay_id, bid) + } + + #[tokio::test] + async fn test_stream_fallback_counts_a_refused_handshake_not_a_204() { + for (status, fallbacks) in [(StatusCode::NO_CONTENT, 0), (StatusCode::NOT_FOUND, 1)] { + let addr = start_status_relay(status).await; + let (relay_id, bid) = get_header_from_stream_relay(addr, 1, B256::ZERO).await; + assert!(bid.is_none()); + assert_eq!( + RELAY_STATUS_CODE + .with_label_values(&[ + status.as_str(), + GET_HEADER_STREAM_ENDPOINT_TAG, + &relay_id + ]) + .get(), + 1, + "handshake answered {status}" + ); + assert_eq!( + RELAY_STREAM_FALLBACK + .with_label_values(&[GET_HEADER_STREAM_ENDPOINT_TAG, &relay_id]) + .get(), + fallbacks, + "handshake answered {status}" + ); + } + } + + // A window whose every bid fails validation records 200 and one failed + // stream. Its updates count every bid it accepted, not only those it held. + #[tokio::test] + async fn test_stream_fallback_counts_a_window_with_no_valid_bid() { + let addr = start_invalid_bid_stream_relay(10).await; + let (relay_id, bid) = get_header_from_stream_relay(addr, 1, B256::ZERO).await; + assert!(bid.is_none()); + + let status = RELAY_STATUS_CODE.with_label_values(&[ + StatusCode::OK.as_str(), + GET_HEADER_STREAM_ENDPOINT_TAG, + &relay_id, + ]); + assert_eq!(status.get(), 1); + let stream = [GET_HEADER_STREAM_ENDPOINT_TAG, relay_id.as_str()]; + assert_eq!(RELAY_STREAM_FALLBACK.with_label_values(&stream).get(), 1); + assert_eq!(RELAY_STREAM_UPDATES.with_label_values(&stream).get_sample_sum(), 10.0); + } + + /// The slot whose start is the fixture bid's timestamp, on Holesky + const FIXTURE_SLOT: u64 = 1_714_953; + + /// The fixture's get_header response bidding `gwei`, and its parent hash + fn fixture_bid_json(gwei: u64) -> (B256, Vec) { + let json = fs::read("../../tests/data/get_header/fulu.json").unwrap(); + let mut json: serde_json::Value = serde_json::from_slice(&json).unwrap(); + let wei = U256::from(gwei) * U256::from(1_000_000_000); + json["data"]["message"]["value"] = wei.to_string().into(); + let parent_hash = json["data"]["message"]["header"]["parent_hash"].as_str().unwrap(); + (parent_hash.parse().unwrap(), serde_json::to_vec(&json).unwrap()) + } + + /// A relay serving `http_body` as its HTTP get_header and `frame` on its + /// stream, both on one port + async fn start_dual_relay(http_body: Vec, frame: Vec) -> SocketAddr { + let http = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let http_addr = http.local_addr().unwrap(); + let relay = axum::Router::new().fallback(move || { + let body = http_body.clone(); + async move { ([(reqwest::header::CONTENT_TYPE, "application/json")], body) } + }); + tokio::spawn(async move { axum::serve(http, relay).await }); + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + while let Ok((mut socket, _)) = listener.accept().await { + let frame = frame.clone(); + tokio::spawn(async move { + let mut head = [0; 2048]; + let n = socket.peek(&mut head).await.unwrap_or_default(); + let head = String::from_utf8_lossy(&head[..n]).to_ascii_lowercase(); + if !head.contains("upgrade: websocket") { + let Ok(mut http) = tokio::net::TcpStream::connect(http_addr).await else { + return; + }; + let _ = tokio::io::copy_bidirectional(&mut socket, &mut http).await; + return; + } + let Ok(mut stream) = accept_async(socket).await else { return }; + let _ = stream.send(Message::Binary(frame.into())).await; + let _ = stream.close(None).await; + }); + } + }); + addr + } + + // A relay's gauges record the better of its HTTP and stream bids + #[tokio::test] + async fn test_relay_gauges_take_the_better_leg() { + for (http_gwei, stream_gwei) in [(60, 50), (50, 60)] { + let (parent_hash, http_bid) = fixture_bid_json(http_gwei); + let (_, stream_bid) = fixture_bid_json(stream_gwei); + // Message type bid, fork fulu + let mut frame = vec![0x01, 6]; + frame.extend(decode_json_payload(&stream_bid).unwrap().data.as_ssz_bytes()); + let addr = start_dual_relay(http_bid, frame).await; + + let (relay_id, bid) = + get_header_from_stream_relay(addr, FIXTURE_SLOT, parent_hash).await; + assert_eq!(*bid.expect("a bid").value(), U256::from(60_000_000_000u64)); + assert_eq!(RELAY_HEADER_VALUE.with_label_values(&[&relay_id]).get(), 60); + assert_eq!(RELAY_LAST_SLOT.with_label_values(&[&relay_id]).get(), FIXTURE_SLOT as i64); + } + } } diff --git a/crates/pbs/src/mev_boost/get_header_ws.rs b/crates/pbs/src/mev_boost/get_header_ws.rs index a91017ec..3fd9bd20 100644 --- a/crates/pbs/src/mev_boost/get_header_ws.rs +++ b/crates/pbs/src/mev_boost/get_header_ws.rs @@ -1,83 +1,28 @@ -//! Streaming get_header over a websocket, for relays configured with +//! get_header over the bid stream, for relays configured with //! `get_header = "stream"`. The stream URL is derived from the relay's own -//! `url`. One connection per get_header call, dropped when the call returns. -//! -//! The request is the handshake itself: slot / parent_hash / pubkey in the -//! path, deadline and timestamp 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 SignedBuilderBid -//! ``` +//! `url`. The handshake carries the same data as the HTTP request: slot / +//! parent_hash / pubkey in the path, deadline and timestamp in headers. -use std::{ - sync::{Arc, OnceLock}, - time::Duration, -}; +use std::time::Duration; use alloy::primitives::utils::format_ether; -use axum::http::{HeaderValue, Request, header::USER_AGENT}; -use cb_common::{ - pbs::{ - ForkName, GetHeaderInfo, GetHeaderResponse, HEADER_START_TIME_UNIX_MS, HEADER_TIMEOUT_MS, - RelayClient, error::PbsError, - }, - utils::utcnow_ms, -}; -use futures::StreamExt; +use cb_common::pbs::{GetHeaderInfo, GetHeaderResponse, RelayClient, error::PbsError}; 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 tokio::time::Instant; use tracing::{debug, info, warn}; use url::Url; -use super::get_header::{RequestInfo, validate_get_header_response}; +use super::get_header::{RequestInfo, decode_ssz_payload, validate_get_header_response}; use crate::{ - constants::{ - GET_HEADER_STREAM_ENDPOINT_TAG, MAX_SIZE_GET_HEADER_RESPONSE, TIMEOUT_ERROR_STATUS, - TRANSPORT_ERROR_STATUS, - }, - metrics::{ - RELAY_LATENCY, RELAY_STATUS_CODE, RELAY_STREAM_CONNECT_LATENCY, - RELAY_STREAM_INVALID_FRAMES, RELAY_STREAM_UPDATES, - }, - mev_boost::get_header::decode_ssz_payload, + bid_stream::{Frame, Held, MAX_HELD_FRAMES, handshake_request, read_bid_stream}, + constants::{GET_HEADER_STREAM_ENDPOINT_TAG, TRANSPORT_ERROR_STATUS}, + metrics::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 { - // TODO @nina: I don't see a point of extending a u8 for supporting older forks - // we could rotate these instead, i.e. 0 becomes Heze, etc - 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, keep the latest bid until the deadline, then -/// validate and return it. +/// Open a stream to the relay, hold its newest bids until the deadline, then +/// return the latest one that passes validation. pub(super) async fn get_header_ws( request_info: &RequestInfo, relay: &RelayClient, @@ -98,126 +43,61 @@ async fn stream_header( timeout_ms: u64, ) -> StreamOutcome { let deadline = Instant::now() + Duration::from_millis(timeout_ms); - let request = match build_handshake_request(request_info, relay, &url, timeout_ms) { + let request = match handshake_request(&url, relay, &request_info.headers) { 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_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<(ForkName, Bytes)> = 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, + let Held { frames, updates, connect_latency, first_frame_latency, invalid_frames } = + match read_bid_stream( + request, + deadline, + relay, + GET_HEADER_STREAM_ENDPOINT_TAG, + MAX_HELD_FRAMES, + Ok, + ) + .await + { + Ok(held) => held, + Err((status, err)) => return (status, Err(err)), }; - match parse_frame(payload) { - Ok((fork, bid)) => { - updates += 1; - if first_bid_latency.is_none() { - first_bid_latency = Some(start_request.elapsed()); - } - latest = Some((fork, bid)); - } + // The latest valid bid stands. Newest first, so the usual case validates once + let start_validate = Instant::now(); + let mut newest_err = None; + let latest_valid = frames.iter().rev().find_map(|frame| { + let validated = decode_ssz_payload(&frame.bid, frame.fork).and_then(|response| { + validate_get_header_response(request_info, relay, &response).map(|()| response) + }); + match validated { + Ok(response) => Some((frame, response)), Err(err) => { - invalid_frames += 1; - if invalid_frames == 1 { - warn!(relay_id = relay.id.as_ref(), %err, "invalid ws frame, skipping"); - } + newest_err.get_or_insert(err); + None } } - } - - drop(stream); - - RELAY_STREAM_UPDATES.with_label_values(&[relay.id.as_str()]).observe(updates as f64); - if invalid_frames > 0 { - RELAY_STREAM_INVALID_FRAMES - .with_label_values(&[relay.id.as_str()]) - .inc_by(invalid_frames as u64); - } + }); + let validate_latency = start_validate.elapsed(); - let Some((fork, bid_bytes)) = latest else { - if let Some(err) = stream_error { - return (TRANSPORT_ERROR_STATUS, Err(err)); + let (Frame { fork, bid }, response) = match (latest_valid, newest_err) { + (Some(valid), None) => valid, + (Some(valid), Some(err)) => { + warn!(relay_id = relay.id.as_ref(), %err, "latest stream bid invalid, an earlier one stands"); + valid + } + (None, Some(err)) => return (StatusCode::OK, Err(err)), + (None, None) => { + debug!(relay_id = relay.id.as_ref(), ?connect_latency, invalid_frames, "no header"); + return (StatusCode::NO_CONTENT, Ok(None)); } - - debug!(relay_id = relay.id.as_ref(), ?connect_latency, invalid_frames, "no header"); - return (StatusCode::NO_CONTENT, Ok(None)); - }; - - if let Some(first_bid_latency) = first_bid_latency { - RELAY_LATENCY - .with_label_values(&[GET_HEADER_STREAM_ENDPOINT_TAG, &relay.id]) - .observe(first_bid_latency.as_secs_f64()); - } - - let response = match decode_ssz_payload(&bid_bytes, fork) { - Ok(response) => response, - Err(err) => return (StatusCode::OK, Err(err)), }; - let start_validate = Instant::now(); - let validated = validate_get_header_response(request_info, relay, &response); - let validate_latency = start_validate.elapsed(); - - if let Err(err) = validated { - return (StatusCode::OK, Err(err)); - } - info!( relay_id = relay.id.as_ref(), - header_size_bytes = bid_bytes.len(), + header_size_bytes = bid.len(), ?connect_latency, - ?first_bid_latency, + first_bid_latency = ?first_frame_latency, ?validate_latency, version = ?fork, value_eth = format_ether(*response.value()), @@ -229,200 +109,3 @@ async fn stream_header( (StatusCode::OK, Ok(Some(response))) } - -/// 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( - request_info: &RequestInfo, - relay: &RelayClient, - 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) = request_info.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(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 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 std::{fs, path::Path}; - - use ssz::Encode; - - 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); - } - - #[test] - fn test_decode_streamed_bid() { - let json_bytes = - fs::read(Path::new("../../tests/data/get_header/fulu.json")).expect("file not found"); - let expected: GetHeaderResponse = - serde_json::from_slice(&json_bytes).expect("failed to decode JSON"); - - let frame = bid_frame(6, &expected.data.as_ssz_bytes()); - let (fork, bid_bytes) = parse_frame(frame).unwrap(); - - let decoded = decode_ssz_payload(&bid_bytes, fork).unwrap(); - assert_eq!(fork, ForkName::Fulu); - assert_eq!(decoded.data, expected.data); - } -} diff --git a/crates/pbs/src/routes/execution_payload_bid.rs b/crates/pbs/src/routes/execution_payload_bid.rs index affdfcf1..31f7afe7 100644 --- a/crates/pbs/src/routes/execution_payload_bid.rs +++ b/crates/pbs/src/routes/execution_payload_bid.rs @@ -7,10 +7,11 @@ use axum::{ response::{IntoResponse, Response}, }; use cb_common::{ + config::GetHeaderTransport, 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::{ @@ -26,6 +27,7 @@ use reqwest::{ use ssz::{Decode, Encode}; use tracing::{debug, error, info, warn}; +use super::execution_payload_bid_ws::get_execution_payload_bid_ws; use crate::{ PbsStateGuard, constants::{ @@ -166,26 +168,78 @@ 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 streams = stream_url.is_some(); + let (http, stream) = match stream_url { + None => ( + send_get_execution_payload_bid(params, body, relay, send_headers, max_timeout_ms).await, + None, + ), + Some(url) => tokio::join!( + send_get_execution_payload_bid( + params, + body.clone(), + relay.clone(), + send_headers.clone(), + max_timeout_ms, + ), + get_execution_payload_bid_ws(&body, &relay, &send_headers, url, max_timeout_ms) + ), + }; + // 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 { - 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 - RELAY_HEADER_VALUE - .with_label_values(&[relay_id.as_str()]) - .set(i64::try_from(relay_bid.bid.data.message.value).unwrap_or_default()); - Ok(Some(relay_bid)) - } - Ok(None) => Ok(None), + // The builder's own 400 and 401 reach the proposer when no bid does. + let http = match http { + Ok(bid) => bid, Err(err) if err.is_timeout() => { error!(err = "Timed Out", %relay_id, timeout_ms = max_timeout_ms); - Ok(None) + None } Err(err) => { error!(err = ?err, %relay_id); - builder_rejection(&err).map_or(Ok(None), Err) + match builder_rejection(&err) { + Some(rejection) if stream.is_none() => return Err(rejection), + _ => None, + } + } + }; + + let selected = select_bid(http, stream); + if streams && let Some((_, transport)) = &selected { + info!(transport = transport.as_str(), "selected bid"); + } + let relay_bid = selected.map(|(relay_bid, _)| relay_bid); + if let Some(relay_bid) = &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 + RELAY_HEADER_VALUE + .with_label_values(&[relay_id.as_str()]) + .set(i64::try_from(relay_bid.bid.data.message.value).unwrap_or_default()); + } + Ok(relay_bid) +} + +/// The higher `value + execution_payment` and the leg it came from, the +/// stream's bid on a tie. Unclamped: the beacon node applies its own cap. +fn select_bid( + http: Option, + stream: Option, +) -> Option<(RelayBid, GetHeaderTransport)> { + let total = |relay_bid: &RelayBid| { + let bid = &relay_bid.bid.data.message; + bid.value.saturating_add(bid.execution_payment) + }; + match (http, stream) { + (Some(http), Some(stream)) if total(&http) > total(&stream) => { + Some((http, GetHeaderTransport::Http)) } + (http, None) => http.map(|bid| (bid, GetHeaderTransport::Http)), + (_, stream) => stream.map(|bid| (bid, GetHeaderTransport::Stream)), } } @@ -267,15 +321,17 @@ async fn send_get_execution_payload_bid( return Ok(None); } - let bid = match content_type { + let relay_bid = match content_type { EncodingType::Json => { - serde_json::from_slice(&response_bytes).map_err(|err| PbsError::JsonDecode { - err, - raw: String::from_utf8_lossy( - &response_bytes[..response_bytes.len().min(MAX_SIZE_DEFAULT)], - ) - .into_owned(), - })? + let bid = + serde_json::from_slice(&response_bytes).map_err(|err| PbsError::JsonDecode { + err, + raw: String::from_utf8_lossy( + &response_bytes[..response_bytes.len().min(MAX_SIZE_DEFAULT)], + ) + .into_owned(), + })?; + RelayBid { bid, body: response_bytes.into(), encoding: EncodingType::Json } } EncodingType::Ssz => { // SSZ requires the fork from Eth-Consensus-Version; its absence is a @@ -285,24 +341,48 @@ 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.into(), fork)? } }; - Ok(Some(RelayBid { bid, body: response_bytes.into(), encoding: content_type })) + Ok(Some(relay_bid)) +} + +pub(super) fn decode_ssz_bid(body: Bytes, fork: ForkName) -> Result { + let data = SignedExecutionPayloadBid::from_ssz_bytes(&body).map_err(|err| { + PbsError::SSZDecode { err: format!("error decoding relay payload: {err:?}"), fork } + })?; + let bid = GetExecutionPayloadBidResponse { version: fork, data, metadata: Default::default() }; + Ok(RelayBid { bid, body, encoding: EncodingType::Ssz }) } #[cfg(test)] mod tests { + use std::{net::SocketAddr, sync::Arc}; + + use alloy::primitives::B256; + use cb_common::{ + config::{PbsModuleConfig, RelayConfig}, + pbs::{BuilderRequestAuth, ExecutionPayloadBid, RelayEntry}, + types::{BlsSecretKey, BlsSignature, Chain}, + }; + use futures::SinkExt; + use lh_types::Slot; + use prometheus::TextEncoder; + use tokio::net::TcpListener; + 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_LATENCY, RELAY_STATUS_CODE, RELAY_STREAM_CONNECT_LATENCY, + RELAY_STREAM_FALLBACK, RELAY_STREAM_INVALID_FRAMES, RELAY_STREAM_UPDATES, + }, + }; #[test] fn test_request_budget_ms() { @@ -355,4 +435,257 @@ 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 } + } + + // (http, stream, served): each bid a `(value, execution_payment)` + #[test] + fn test_select_bid() { + for (http, stream, served) in [ + // The HTTP bid's execution payment counts as much as the stream's + (Some((10, 60)), Some((50, 0)), Some((GetHeaderTransport::Http, (10, 60)))), + // The stream wins a tie + (Some((50, 0)), Some((10, 40)), Some((GetHeaderTransport::Stream, (10, 40)))), + // A total past u64::MAX saturates rather than wrapping to a low one + (Some((u64::MAX, 1)), Some((1, 0)), Some((GetHeaderTransport::Http, (u64::MAX, 1)))), + (None, Some((1, 0)), Some((GetHeaderTransport::Stream, (1, 0)))), + (Some((1, 0)), None, Some((GetHeaderTransport::Http, (1, 0)))), + (None, None, None), + ] { + let selected = select_bid( + http.map(|(v, p)| relay_bid(v, p)), + stream.map(|(v, p)| relay_bid(v, p)), + ) + .map(|(relay_bid, transport)| { + let bid = &relay_bid.bid.data.message; + (transport, (bid.value, bid.execution_payment)) + }); + assert_eq!(selected, served, "http {http:?}, stream {stream:?}"); + } + } + + /// A relay that answers every request, the stream handshake included, with + /// `status` and no body + async fn start_status_relay(status: StatusCode) -> SocketAddr { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let relay = axum::Router::new().fallback(move || async move { status }); + tokio::spawn(async move { axum::serve(listener, relay).await }); + addr + } + + /// 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 { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + while let Ok((socket, _)) = listener.accept().await { + let messages = messages.clone(); + tokio::spawn(async move { + 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 + } + + /// A relay that accepts the connection and never answers the handshake + async fn start_silent_relay() -> SocketAddr { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + let mut held = Vec::new(); + while let Ok((socket, _)) = listener.accept().await { + held.push(socket); + } + }); + addr + } + + 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 failed_streams(relay_id: &str) -> u64 { + RELAY_STREAM_FALLBACK.with_label_values(&[STREAM, relay_id]).get() + } + + // A handshake answered 204 is no bid. A refused handshake, one that runs + // out the window, a stream that breaks before a bid, and a frame over the + // size cap are failed streams. + #[tokio::test] + async fn test_epbs_stream_failures_by_status() { + let oversized = Message::Binary(vec![0; MAX_SIZE_GET_HEADER_RESPONSE + 1].into()); + for (relay, status, failed) in [ + (start_status_relay(StatusCode::NO_CONTENT).await, StatusCode::NO_CONTENT, 0), + (start_status_relay(StatusCode::NOT_FOUND).await, StatusCode::NOT_FOUND, 1), + (start_silent_relay().await, TIMEOUT_ERROR_STATUS, 1), + (start_stream_relay(vec![], ZERO, false).await, TRANSPORT_ERROR_STATUS, 1), + (start_stream_relay(vec![oversized], ZERO, true).await, TRANSPORT_ERROR_STATUS, 1), + ] { + let (relay_id, bid) = bid_from_stream_relay(relay, 300).await; + assert!(bid.is_none(), "{status}"); + assert_eq!(status_count(status, &relay_id), 1, "{status}"); + assert_eq!(failed_streams(&relay_id), failed, "{status}"); + } + } + + // 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 = 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!(failed_streams(&relay_id), 0); + let stream = [STREAM, relay_id.as_str()]; + assert_eq!( + RELAY_STREAM_INVALID_FRAMES.with_label_values(&stream).get(), + n_frames as u64 + ); + assert_eq!(RELAY_STREAM_UPDATES.with_label_values(&stream).get_sample_sum(), 0.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. + #[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 = start_stream_relay(messages, Duration::from_millis(150), true).await; + let (relay_id, bid) = bid_from_stream_relay(relay, 2_000).await; + 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!(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 stream = [STREAM, relay_id.as_str()]; + assert_eq!(RELAY_STREAM_UPDATES.with_label_values(&stream).get_sample_sum(), 2.0); + assert_eq!(RELAY_STREAM_CONNECT_LATENCY.with_label_values(&stream).get_sample_count(), 1); + // The last bid arrives 450 ms in + let latency = RELAY_LATENCY.with_label_values(&stream); + assert_eq!(latency.get_sample_count(), 1); + assert!(latency.get_sample_sum() < 0.3, "{}", latency.get_sample_sum()); + } + + // Dashboards and the docs select the stream series by label 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, 1_000).await; + let (failed_id, _) = + bid_from_stream_relay(start_status_relay(StatusCode::NOT_FOUND).await, 300).await; + + let scrape = TextEncoder::new().encode_to_string(&PBS_METRICS_REGISTRY.gather()).unwrap(); + for (series, relay_id) in [ + ("relay_stream_connect_latency_count", &relay_id), + ("relay_stream_updates_sum", &relay_id), + ("relay_stream_invalid_frames_total", &relay_id), + ("relay_stream_fallback_total", &failed_id), + ] { + let series = format!(r#"{series}{{endpoint="{STREAM}",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..fb5987df --- /dev/null +++ b/crates/pbs/src/routes/execution_payload_bid_ws.rs @@ -0,0 +1,98 @@ +//! The ePBS bid over the bid stream of a relay with `get_header = "stream"`. +//! Each frame carries an SSZ `SignedExecutionPayloadBid`. + +use std::time::Duration; + +use axum::http::{HeaderMap, HeaderValue}; +use base64::{Engine, prelude::BASE64_STANDARD}; +use cb_common::{ + pbs::{HEADER_REQUEST_AUTH, RelayClient, error::PbsError}, + wire::CONSENSUS_VERSION_HEADER, +}; +use reqwest::StatusCode; +use tokio::time::Instant; +use tracing::debug; +use url::Url; + +use super::execution_payload_bid::{RelayBid, decode_ssz_bid}; +use crate::{ + bid_stream::{Frame, Held, handshake_request, read_bid_stream, record_stream_failure}, + constants::{GET_EXECUTION_PAYLOAD_BID_STREAM_ENDPOINT_TAG, TRANSPORT_ERROR_STATUS}, + metrics::RELAY_STATUS_CODE, +}; + +type StreamOutcome = (StatusCode, Result, PbsError>); + +/// The relay's latest bid at the deadline. A failed stream is no bid. +pub(super) async fn get_execution_payload_bid_ws( + auth_ssz: &[u8], + relay: &RelayClient, + send_headers: &HeaderMap, + url: Url, + timeout_ms: u64, +) -> Option { + let (status, res) = + stream_execution_payload_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.unwrap_or_else(|err| { + record_stream_failure(endpoint, &relay.id, &err); + None + }) +} + +async fn stream_execution_payload_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 mut request = match handshake_request(&url, relay, send_headers) { + Ok(request) => request, + Err(err) => return (TRANSPORT_ERROR_STATUS, Err(err)), + }; + let headers = request.headers_mut(); + if let Some(fork) = send_headers.get(CONSENSUS_VERSION_HEADER) { + headers.insert(CONSENSUS_VERSION_HEADER, fork.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); + + // Each frame is decoded on arrival, so the newest is the latest bid that + // decodes and the only one to hold + let parse = |frame: Frame| decode_ssz_bid(frame.bid, frame.fork); + let Held { mut frames, updates, connect_latency, first_frame_latency, invalid_frames } = + match read_bid_stream( + request, + deadline, + relay, + GET_EXECUTION_PAYLOAD_BID_STREAM_ENDPOINT_TAG, + 1, + parse, + ) + .await + { + Ok(held) => held, + Err((status, err)) => return (status, Err(err)), + }; + + let Some(bid) = frames.pop_back() else { + debug!(relay_id = relay.id.as_ref(), ?connect_latency, invalid_frames, "no bid"); + return (StatusCode::NO_CONTENT, Ok(None)); + }; + + debug!( + relay_id = relay.id.as_ref(), + ?connect_latency, + first_bid_latency = ?first_frame_latency, + updates, + invalid_frames, + "received new bid from ws stream" + ); + + (StatusCode::OK, Ok(Some(bid))) +} 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..c5d48616 100644 --- a/docs/docs/get_started/configuration.md +++ b/docs/docs/get_started/configuration.md @@ -75,7 +75,7 @@ Each `[[relays]]` entry supports, besides `id` and `url`: - `headers`: custom headers sent with every request to this relay, which is how a relay API key is supplied. See [Relay API keys](#relay-api-keys). - `get_params`: optional GET parameters to add to each request URL for this relay. -- `get_header`: how bids are fetched from this relay, either `"http"` (one request per `get_header`) or `"stream"` (a websocket stream of bid updates, for relays that support it). Default: `"http"`. See [Bid streaming](#bid-streaming). +- `get_header`: how bids are fetched from this relay: `"http"`, one request per `get_header`, or `"stream"`, a websocket stream of bid updates for relays that support it. (unreleased, from v0.12.0-rc1) A streaming relay also gets the HTTP request. Default: `"http"`; details under [Bid streaming](#bid-streaming). - `enable_timing_games`: whether to enable timing games for this relay, as tuned by `target_first_request_ms` and `frequency_get_header_ms`. If neither of those is set, this flag has no effect. Advanced users only: misconfiguration can result in e.g. fetching a lower header value or missing a slot (caveats and worked examples in the annotated config example). Default: `false`. - `target_first_request_ms`: target time in the slot, in milliseconds, at which to send the first `get_header` request. - `frequency_get_header_ms`: frequency, in milliseconds, at which to send `get_header` requests. @@ -110,7 +110,19 @@ Under Docker, `commit-boost init` mounts each `file` path into the container at ### Bid streaming -`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. +:::info Unreleased +From v0.12.0-rc1, compared with v0.11.0: +- The relay's HTTP request runs alongside every stream. In v0.11.0 it runs only when the handshake fails with time left. +- The stream serves its latest valid bid. In v0.11.0 only the last bid is validated. +- A `204` answer to the handshake is no bid. In v0.11.0 it records `556` and retries over HTTP. +- `X-Timeout-Ms` on the handshake leaves out half the TCP connect time. In v0.11.0 it is the whole window. +- A failed stream logs `stream failed`. In v0.11.0 it logs `stream failed, falling back to http get_header`, and only when it retries. +- A relay also listed with `get_header = "http"`, as the v0.11.0 docs advised, gets two HTTP requests per `get_header`, and after the Gloas fork only the first of the two serves ePBS; remove the `"http"` entry. +::: + +`get_header = "stream"` adds a websocket stream of bid updates to the relay's per-slot HTTP request. 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 @@ -120,44 +132,29 @@ For each `get_header` call PBS dials: ws(s):///eth/v1/builder/header_stream/{slot}/{parent_hash}/{pubkey} ``` -The scheme and host come from the relay's `url`: `https` becomes `wss`, `http` becomes `ws`, any other scheme fails at startup. The pubkey is dropped from the userinfo and `get_params` are appended as usual. +The scheme and host come from the relay's `url`: `https` becomes `wss`, `http` becomes `ws`, any other scheme fails at startup. The pubkey is dropped from the userinfo and `get_params` are appended as usual. The stream connects directly, ignoring `HTTP_PROXY`, `HTTPS_PROXY` and `ALL_PROXY`. -The handshake carries what the HTTP request would: slot, parent hash and pubkey in the path, plus `Date-Milliseconds`, `X-Timeout-Ms`, `X-CommitBoost-Version` and every configured header. +The handshake carries what the HTTP request would: slot, parent hash and pubkey in the path, plus `Date-Milliseconds`, `X-Timeout-Ms`, `X-CommitBoost-Version` and every configured header. `X-Timeout-Ms` is the time left in the bid window minus half the TCP connect time, so the relay's last update arrives before the window closes. The stream needs the same API key as the HTTP path, sent on the handshake with the rest of `headers`. See [Relay API keys](#relay-api-keys). #### Bid window -Once connected, the relay sends one binary frame per bid update. PBS keeps the most recent update, and validates and returns it when the window ends or the relay closes the stream. Validation is the same as on the HTTP path, so `skip_sigverify`, `min_bid_eth` and `extra_validation_enabled` apply unchanged. Frames that do not parse as a bid are skipped and counted. +Once connected, the relay sends one binary frame per bid update. PBS keeps the last 8 bids and, when the window ends or the relay closes the stream, returns the newest one that passes validation. Validation is the same as on the HTTP path, so `skip_sigverify`, `min_bid_eth` and `extra_validation_enabled` apply unchanged. Frames whose message type or fork byte cannot be read are skipped and counted. The window is the deadline the HTTP path already computes: `timeout_get_header_ms`, capped by the time left until `late_in_slot_time_ms`, and capped again by the CL's `X-Timeout-Ms` if it sends one. An HTTP `get_header` normally returns early, while a stream is held to the end, so on a streaming relay `timeout_get_header_ms` sets how long PBS holds the CL's request. A CL request arriving at or after `late_in_slot_time_ms` still skips relays entirely and forces local building. -The timing-game options have no effect on a stream. They schedule repeated HTTP requests, and the stream already delivers every update the relay produces. They do apply to the HTTP fallback. - -#### Fallback - -If the handshake fails, PBS falls back to a plain HTTP `get_header` at the relay's normal URL with whatever time is left in the window, and counts `cb_pbs_relay_stream_fallback_total`. That covers a refused connection, a DNS or TLS failure, and a relay that rejects the upgrade with an HTTP response. +The timing-game options have no effect on the stream, which already delivers every update the relay produces. They apply to the HTTP request that runs alongside it. -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. +#### HTTP request -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. +PBS sends the relay its normal HTTP `get_header` alongside the stream. The stream's bid and the HTTP bid are separate candidates and the better one wins, so a failed stream leaves the HTTP bid in the auction. The `auction winner` log line names the winning `transport`. -```toml -[[relays]] -id = "relay-1-stream" -url = "https://0xa1cec75a3f0661e99299274182938151e8433c61a19222347ea1313d839229cb4ce4e3e5aa2bdeb71c8fcf1b084963c2@relay-1.xyz" -get_header = "stream" -headers = { X-Api-Key = "..." } - -id = "relay-1-http" -url = "https://0xa1cec75a3f0661e99299274182938151e8433c61a19222347ea1313d839229cb4ce4e3e5aa2bdeb71c8fcf1b084963c2@relay-1.xyz" -get_header = "http" -headers = { X-Api-Key = "..." } -``` +A stream fails when it cannot connect, when the handshake times out or the relay refuses it, when the connection drops before any bid arrives, or when none of the bids PBS kept passes validation. Commit-Boost logs each failed stream as `stream failed` with the reason, and counts it in `cb_pbs_relay_stream_fallback_total`. A drop after a bid arrived is not a failure: PBS validates the bids it kept as usual. A `204` answer to the handshake means the relay has no bid. -#### Metrics +#### Stream metrics -Stream and fallback outcomes are recorded separately. `cb_pbs_relay_status_code_total` and `cb_pbs_relay_latency` carry `endpoint="get_header_stream"` for the stream and `endpoint="get_header"` for the fallback, and four `cb_pbs_relay_stream_*` series cover handshake latency, updates per window, unparseable frames and fallbacks. See [Metrics > Bid stream](./running/metrics.md#bid-stream). +Stream and HTTP outcomes are recorded separately. `cb_pbs_relay_status_code_total` and `cb_pbs_relay_latency` carry `endpoint="get_header_stream"` for the stream and `endpoint="get_header"` for the HTTP request, and four `cb_pbs_relay_stream_*` series cover handshake latency, updates per window, unparseable frames and failed streams; [Metrics > Bid stream](./running/metrics.md#bid-stream) says what each value means. ### SSZ support diff --git a/docs/docs/get_started/epbs.md b/docs/docs/get_started/epbs.md index 1c14de6c..b763c0da 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 also reads its bids from a [stream](#bid-streaming). ## How a bid request reaches a builder @@ -92,16 +92,29 @@ 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 uses the HTTP bid. 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 sends the usual HTTP request and opens the stream at the same time. It returns the HTTP bid or the latest streamed bid that decodes, whichever has the higher `value + execution_payment`, and the streamed bid on a tie. Commit-Boost does not apply the key's `max_execution_payment` to this choice; the beacon node applies it afterwards. If the stream fails, the HTTP bid still reaches the beacon node. A builder [outside your config](#builders-outside-your-config) is always asked over HTTP. + +Streaming makes bid requests slower. Commit-Boost replies to the beacon node once the builder's HTTP response is in and the stream has ended: when the builder closes it, when it fails, or at the beacon node's deadline less `proposer_deadline_buffer_ms`. An HTTP builder is answered as soon as it 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](./configuration.md#connection), without your proxy settings, while the HTTP request 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`. | 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 | | 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"`, except one Commit-Boost [refuses to dial](#builders-outside-your-config), which gets no request | | How fast are builders? | `cb_pbs_relay_latency`, by `relay_id` | -| 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 | +| 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, any other code a failed stream: `555` the deadline ran out before the handshake completed, `556` a connection that failed or dropped before a bid, or the builder's own refusal such as `404`. When Commit-Boost returns a streaming builder's bid, it also logs `selected bid`, with `transport` set to `stream` or `http` for where that bid came from | +| 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 the bid Commit-Boost returns for each builder; with a streaming builder, whichever of its two bids it chose | ## Troubleshooting @@ -110,7 +123,9 @@ 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 hostname nor an `http(s)` URL, 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). A key without builder config sends Commit-Boost's own hostname, which usually does | 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_stream_fallback_total{endpoint="get_execution_payload_bid_stream"}` rises | The builder does not serve the ePBS bid stream. Its HTTP bid still reaches the beacon node | 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 before the deadline, for example because your network allows only proxied connections, which the stream does not use. The HTTP bid still reaches the beacon node | Keep the builder on `get_header = "http"`, 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` | | The signed block gets `415` | The beacon node sends it as JSON | Configure the beacon node to send SSZ | | Commit-Boost logs "no builder accepted the signed beacon block" | The winning bid came from a builder outside your config, which gets the block over gossip, or your builders did not accept it. The beacon node gossips the block either way | If the bid came from a configured builder, check its answer in `cb_pbs_relay_status_code_total{endpoint="submit_signed_beacon_block"}` | -| 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 4db35ea4..325c1850 100644 --- a/docs/docs/get_started/running/metrics-catalog.md +++ b/docs/docs/get_started/running/metrics-catalog.md @@ -14,14 +14,14 @@ 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), 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 (over HTTP a timeout, DNS or connection failure; on a stream, the bid window ran out before the handshake completed), and `"556"` for a bid-stream transport failure (connect failed, the stream broke before a bid, or the handshake was answered with a `2xx` other than `204`). (unreleased, from v0.12.0-rc1) A `204` answer to the handshake records `204`. 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 the two stream endpoints 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_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 a streaming relay, the better of its HTTP and stream bids. For ePBS, the returned bid's `value`, without its execution payment. | +| `cb_pbs_relay_stream_connect_latency` | Histogram | `endpoint` (unreleased, from v0.12.0-rc1), `relay_id` | Websocket handshake latency in seconds, for a relay with `get_header = "stream"`. Observed once per successful handshake; a handshake that fails, runs out the bid window, or is answered `204` records nothing here. Custom buckets from 5 ms to 1 s. Endpoint values: `get_header_stream`, `get_execution_payload_bid_stream`. | +| `cb_pbs_relay_stream_updates` | Histogram | `endpoint` (unreleased, from v0.12.0-rc1), `relay_id` | Bid updates accepted on arrival per stream: on `get_header_stream`, every frame whose message type and fork byte can be read; on `get_execution_payload_bid_stream`, every bid that decodes. Observed once per stream that connected, including streams that received nothing; a handshake that fails, times out or is answered `204` records nothing. Buckets: `0, 1, 2, 3, 5, 10, 20, 50, 100`. Endpoint values: `get_header_stream`, `get_execution_payload_bid_stream`. | +| `cb_pbs_relay_stream_invalid_frames_total` | Counter | `endpoint` (unreleased, from v0.12.0-rc1), `relay_id` | Websocket frames whose message type or fork byte cannot be read; on the ePBS stream, also bids that fail to decode. Incremented only for windows that saw at least one, so the series is absent while zero. Endpoint values: `get_header_stream`, `get_execution_payload_bid_stream`. | +| `cb_pbs_relay_stream_fallback_total` | Counter | `endpoint` (unreleased, from v0.12.0-rc1), `relay_id` | (unreleased, from v0.12.0-rc1; in v0.11.0, handshake failures retried over HTTP) Bid streams that failed while the relay's HTTP request ran alongside: a connection or handshake that failed or timed out, a stream that broke before a bid, or, on `get_header_stream`, a window in which every held bid was invalid. A `204` answer to the handshake is no bid and does not count. Counts every failed stream, whether or not the HTTP request returned a bid; the HTTP result is recorded under `endpoint="get_header"` (or `get_execution_payload_bid`). Endpoint values: `get_header_stream`, `get_execution_payload_bid_stream`. | | `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..1d5efdb6 100644 --- a/docs/docs/get_started/running/metrics.md +++ b/docs/docs/get_started/running/metrics.md @@ -112,13 +112,25 @@ Once Grafana is running, you can [import](https://grafana.com/docs/grafana/lates ## Bid stream -For a relay with `get_header = "stream"` (see [Configuration > Bid streaming](../configuration.md#bid-streaming)), every series below is labelled by `relay_id`. The stream's outcome and time-to-first-bid share the two general relay series under `endpoint="get_header_stream"`, and the HTTP fallback keeps `endpoint="get_header"`. +:::info Unreleased +From v0.12.0-rc1, compared with v0.11.0: +- The `cb_pbs_relay_stream_*` series carry an `endpoint` label. In v0.11.0 they carry `relay_id` only. +- `cb_pbs_relay_stream_fallback_total` counts every failed stream. In v0.11.0 it counts handshake failures retried over HTTP. +- A handshake answered `204` records `204`. In v0.11.0 it records `556`. +- A streaming relay's HTTP request records under `endpoint="get_header"` on every request. In v0.11.0 it records only on a fallback. +::: + +For a relay with `get_header = "stream"` (see [Configuration > Bid streaming](../configuration.md#bid-streaming)), every series below is labeled by `relay_id` and `endpoint`. The stream's outcome and time-to-first-bid share the two general relay series under `endpoint="get_header_stream"`, and the HTTP request that runs alongside it keeps `endpoint="get_header"`. | Question | Series | |---|---| -| Is the stream serving bids? | `cb_pbs_relay_status_code_total{endpoint="get_header_stream"}`: `200` a bid was delivered, `204` connected but no bid before the deadline, `555` the bid window ran out during the handshake, `556` a transport error (connect failed, the stream broke mid-window, or the handshake was answered with a success code instead of the upgrade), any other code the relay's own refusal of the upgrade | +| Is the stream serving bids? | `cb_pbs_relay_status_code_total{endpoint="get_header_stream"}`: `200` a bid was delivered, `204` no bid (the handshake was answered `204`, or the stream ended or the deadline passed without one), `555` the bid window ran out during the handshake, `556` a transport error (connect failed, the stream broke before a bid, or the handshake was answered with a `2xx` other than `204` instead of the upgrade), any other code the relay's own refusal of the upgrade | | How fast does the first bid arrive? | `cb_pbs_relay_latency{endpoint="get_header_stream"}` | -| Is it falling back to HTTP? | `cb_pbs_relay_stream_fallback_total`: handshake failures that had bid window left to retry over HTTP. A steady rate means the relay is refusing or dropping the upgrade. The fallback's own results are under `endpoint="get_header"` | +| Is the stream failing? | `cb_pbs_relay_stream_fallback_total{endpoint="get_header_stream"}`: [failed streams](../configuration.md#http-request), counted whether or not the HTTP request returned a bid. For a steady rate, the status series above and the `stream failed` log line give the reason | | 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 | +| Is it actually streaming? | `cb_pbs_relay_stream_updates`: bids received per request. On `get_header_stream` every frame with a readable message type and fork byte counts, before its bid is checked. A healthy relay sends several; at most one per request means the stream connects but does not stream | +| Are the frames usable? | `cb_pbs_relay_stream_invalid_frames_total`: frames dropped on arrival because their message type or fork byte cannot be read, absent while zero. A bid that fails decoding or validation is not counted here | + +When the newest bid fails validation and an earlier one is served, the failure is logged, not counted. A window in which every kept bid fails counts as `200` on the status series, the same as an invalid bid over HTTP, and as one failed stream. If the beacon node drops the request before the deadline, no status is recorded. + +The [ePBS bid stream](../epbs.md#bid-streaming) (unreleased, from v0.12.0-rc1) reports the same series under `endpoint="get_execution_payload_bid_stream"`, and its HTTP request under `endpoint="get_execution_payload_bid"`. It decodes each bid as it arrives and does not validate it, so `cb_pbs_relay_stream_updates` counts only bids that decode, and `cb_pbs_relay_stream_invalid_frames_total` also counts those that do not. A stream that ends with only undecodable bids is no bid (`204`), not a failed stream; if its connection drops instead, it records `556` and a failed stream. diff --git a/tests/Cargo.toml b/tests/Cargo.toml index 77206686..6e7c876a 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, features = ["testing-flags"] } cb-signer.workspace = true diff --git a/tests/src/mock_relay.rs b/tests/src/mock_relay.rs index c1ad4b81..0a650512 100644 --- a/tests/src/mock_relay.rs +++ b/tests/src/mock_relay.rs @@ -328,6 +328,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 { @@ -394,6 +398,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)>, @@ -453,23 +483,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..8bb28f9a 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,47 +11,73 @@ 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}, + utils::utcnow_ms, + wire::CONSENSUS_VERSION_HEADER, }; use futures::SinkExt; use ssz::Encode; -use tokio::net::{TcpListener, TcpStream}; +use tokio::{ + net::{TcpListener, TcpSocket, TcpStream}, + sync::oneshot, + time::timeout, +}; 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, + /// When the relay accepted the connection, unix ms + pub accepted_at_ms: u64, } pub struct MockWsRelayState { pub chain: Chain, pub signer: BlsSecretKey, - /// One frame pushed per value, in order. The last one is what PBS must - /// return. + /// One frame pushed per value, in order 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 +86,20 @@ 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, + /// Make this many of the last bids fail validation: signed for another + /// parent hash, or on the ePBS stream cut short so they do not decode + invalid_last_bids: usize, + /// Wait this long before answering the handshake, as a relay whose + /// handshake outlasts PBS's budget + handshake_delay: Duration, + /// Drop the connection after the last bid, with no close frame, as a + /// stream that breaks + abort_after_bids: bool, + /// End the stream at Date-Milliseconds + X-Timeout-Ms, as a relay honouring + /// the timing headers does: the last bid goes out then, the earlier ones + /// `update_interval` apart before it, and each reaches PBS this much later + ends_at_timeout: Option, + handshake_attempts: AtomicU64, received_connections: AtomicU64, last_request: Mutex>, } @@ -69,9 +110,16 @@ 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_delay: Duration::ZERO, + abort_after_bids: false, + ends_at_timeout: None, + handshake_attempts: AtomicU64::new(0), received_connections: AtomicU64::new(0), last_request: Mutex::new(None), } @@ -81,6 +129,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 +149,28 @@ impl MockWsRelayState { Self { unknown_frames: true, ..self } } + pub fn with_invalid_last_bids(self, invalid_last_bids: usize) -> Self { + Self { invalid_last_bids, ..self } + } + + pub fn with_handshake_delay(self, handshake_delay: Duration) -> Self { + Self { handshake_delay, ..self } + } + + pub fn abort_after_bids(self) -> Self { + Self { abort_after_bids: true, ..self } + } + + pub fn ends_at_timeout(self, downlink_latency: Duration) -> Self { + Self { ends_at_timeout: Some(downlink_latency), ..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 +198,48 @@ 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<()> { + let accepted_at_ms = utcnow_ms(); + state.handshake_attempts.fetch_add(1, Ordering::Relaxed); + if !state.handshake_delay.is_zero() { + tokio::time::sleep(state.handshake_delay).await; + } let mut request = None; let mut ws = accept_hdr_async(stream, |req: &Request, res: Response| { - request = parse_request(req); + request = parse_request(req, accepted_at_ms); + 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 stream_end_ms = match state.ends_at_timeout { + Some(_) => Some( + request + .start_time_ms + .zip(request.timeout_ms) + .map(|(start, timeout)| start + timeout) + .ok_or_else(|| eyre::eyre!("no timing headers"))?, + ), + None => None, + }; + + 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 let (Some(end_ms), Some(latency)) = (stream_end_ms, state.ends_at_timeout) { + let sent_ms = + end_ms - state.update_interval.as_millis() as u64 * (n_bids - 1 - i) as u64; + let arrives_ms = sent_ms + latency.as_millis() as u64; + tokio::time::sleep(Duration::from_millis(arrives_ms.saturating_sub(utcnow_ms()))).await; + } + 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,23 +247,19 @@ 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() { + if stream_end_ms.is_none() && !state.update_interval.is_zero() { tokio::time::sleep(state.update_interval).await; } } + if state.abort_after_bids { + return Ok(()); + } + if state.hold_open { // Never resolves: PBS has to cut the stream off at its own deadline std::future::pending::<()>().await; @@ -166,27 +270,164 @@ 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 parent_hash = if invalid { B256::repeat_byte(0xee) } else { request.parent_hash }; + let bid = mock_signed_builder_bid( + state.chain, + &state.signer, + request.slot, + 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. -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('/'); +/// `/{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, accepted_at_ms: u64) -> Option { + 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), + accepted_at_ms, }) } fn header(req: &Request, name: &str) -> Option { req.headers().get(name)?.to_str().ok().map(ToOwned::to_owned) } + +/// Fills the accept queue until `drain` fires, so a connect waits a second for +/// its SYN retransmit; a delayed accept would not slow it. Returns the port. +pub async fn start_mock_ws_relay_with_full_backlog( + state: Arc, + drain: oneshot::Receiver<()>, +) -> eyre::Result { + let socket = TcpSocket::new_v4()?; + socket.bind(([127, 0, 0, 1], 0).into())?; + let listener = socket.listen(1)?; + let addr = listener.local_addr()?; + + // How many connections the queue admits is up to the OS: fill it until one + // does not complete + let mut fillers = Vec::new(); + while let Ok(filler) = timeout(Duration::from_millis(200), TcpStream::connect(addr)).await { + fillers.push(filler?); + eyre::ensure!(fillers.len() < 16, "accept queue never filled"); + } + + tokio::spawn(async move { + let _ = drain.await; + for _ in &fillers { + listener.accept().await?; + } + drop(fillers); + start_mock_ws_relay_service(state, listener).await + }); + + Ok(addr.port()) +} + +/// 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..70632917 --- /dev/null +++ b/tests/tests/pbs_get_execution_payload_bid_ws.rs @@ -0,0 +1,193 @@ +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::{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 bidding `http` gwei over HTTP and streaming `stream_bids`, each a +/// `(value, execution_payment)` in gwei +fn relay_states(http: u64, stream_bids: Vec<(u64, u64)>) -> (MockRelayState, MockWsRelayState) { + let signer = random_secret(); + let http_state = + MockRelayState::new(Chain::Hoodi, signer.clone()).with_trustless_bid_gwei(http); + (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_race( + (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 the HTTP request raced alongside it supplies the bid. +#[tokio::test] +async fn test_get_execution_payload_bid_ws_no_stream_route_http_bid_serves() -> Result<()> { + let (http_state, ws_state) = relay_states(42, vec![(90, 0)]); + let (validator, http_state, ws_state) = + start_race((http_state, ws_state.without_epbs_stream())).await?; + + let (code, res) = get_bid(&validator).await?; + assert_eq!(code, StatusCode::OK); + assert_bid(res, 42, 0); + assert_eq!(http_state.received_execution_payload_bid(), 1); + assert_eq!(ws_state.handshake_attempts(), 1); + assert_eq!(ws_state.received_connections(), 0, "the handshake never completed"); + Ok(()) +} + +/// With no bid from the stream, the builder's HTTP 401 reaches the beacon node. +#[tokio::test] +async fn test_get_execution_payload_bid_ws_no_stream_bid_http_rejection_passes() -> Result<()> { + let (http_state, ws_state) = relay_states(10, vec![(50, 0)]); + let (validator, http_state, _) = + start_race((http_state, ws_state.without_epbs_stream())).await?; + http_state.set_response_override(StatusCode::UNAUTHORIZED); + + let (code, _) = get_bid(&validator).await?; + assert_eq!(code, StatusCode::UNAUTHORIZED); + Ok(()) +} + +/// The builder's HTTP 401 does not displace the stream's bid. +#[tokio::test] +async fn test_get_execution_payload_bid_ws_http_rejection_stream_bid_serves() -> Result<()> { + let (validator, http_state, _) = start_race(relay_states(10, vec![(50, 0)])).await?; + http_state.set_response_override(StatusCode::UNAUTHORIZED); + + 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(), 1); + Ok(()) +} + +/// An HTTP request that runs out the window does not displace the stream's bid. +#[tokio::test] +async fn test_get_execution_payload_bid_ws_http_timeout_stream_bid_serves() -> Result<()> { + let (http_state, ws_state) = relay_states(10, vec![(50, 0)]); + let (validator, ..) = start_race((http_state.with_bid_delay_ms(BUDGET_MS), ws_state)).await?; + + let (code, res) = get_bid(&validator).await?; + assert_eq!(code, StatusCode::OK); + assert_bid(res, 50, 0); + Ok(()) +} + +/// The stream and the HTTP request are separate candidates: the higher value + +/// execution_payment wins, the stream's bid on a tie, and the stream's +/// candidate is its latest bid, not its highest. +#[tokio::test] +async fn test_get_execution_payload_bid_ws_higher_total_wins() -> Result<()> { + for (http, stream_bids, served) in + [(60, vec![(70, 0), (50, 0)], (60, 0)), (50, vec![(10, 40)], (10, 40))] + { + let (validator, http_state, ws_state) = + start_race(relay_states(http, stream_bids.clone())).await?; + + let (code, res) = get_bid(&validator).await?; + assert_eq!(code, StatusCode::OK, "http {http}, stream {stream_bids:?}"); + assert_bid(res, served.0, served.1); + assert_eq!(http_state.received_execution_payload_bid(), 1); + 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(10, vec![(50, 0), (90, 0)]); + let (validator, ..) = start_race((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(10, vec![(20, 0)]); + let (_, ws_state, relay) = + start_mock_dual_relay(http_state.with_bid_delay_ms(400), 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 (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" + ); + assert_eq!(request.api_key.as_deref(), Some(API_KEY)); + assert!(request.user_agent.is_some_and(|ua| ua.contains("commit-boost"))); + Ok(()) +} diff --git a/tests/tests/pbs_get_header_ws.rs b/tests/tests/pbs_get_header_ws.rs index e132a70d..e9fcd4b6 100644 --- a/tests/tests/pbs_get_header_ws.rs +++ b/tests/tests/pbs_get_header_ws.rs @@ -10,14 +10,17 @@ use cb_common::{ signature::sign_builder_root, signer::random_secret, types::{BlsPublicKeyBytes, BlsSecretKey, Chain, KnownChain}, - utils::timestamp_of_slot_start_sec, + utils::{timestamp_of_slot_start_sec, utcnow_ms}, wire::EncodingType, }; use cb_pbs::{DefaultBuilderApi, PbsService, PbsState}; use cb_tests::{ mock_relay::{MockRelayState, start_mock_relay_service_with_listener}, mock_validator::MockValidator, - mock_ws_relay::{MockWsRelayState, start_mock_ws_relay_service}, + mock_ws_relay::{ + MockWsRelayState, start_mock_dual_relay, start_mock_ws_relay_service, + start_mock_ws_relay_with_full_backlog, + }, utils::{ API_KEY, generate_mock_relay, generate_mock_stream_relay, generate_mock_stream_relay_with_timing_games, get_free_listener, get_pbs_config, @@ -27,6 +30,7 @@ use cb_tests::{ use eyre::Result; use lh_types::ForkName; use reqwest::StatusCode; +use tokio::sync::oneshot; use tree_hash::TreeHash; fn request_slot() -> u64 { @@ -61,8 +65,7 @@ async fn start_pbs( let config = to_pbs_config(chain, pbs_config, relays); let state = PbsState::new(config, PathBuf::new()); - drop(listener); - tokio::spawn(PbsService::run::<(), DefaultBuilderApi>(state)); + tokio::spawn(PbsService::run_with_listener::<(), DefaultBuilderApi>(state, listener)); // leave some time to start servers tokio::time::sleep(Duration::from_millis(100)).await; @@ -132,6 +135,39 @@ async fn test_get_header_ws_returns_latest_bid() -> Result<()> { Ok(()) } +/// Bids that fail validation do not displace the valid one before them, while +/// it is among the newest 8 the stream holds +#[tokio::test] +async fn test_get_header_ws_returns_latest_valid_bid() -> Result<()> { + setup_test_env(); + let signer = random_secret(); + let chain = Chain::Hoodi; + + for (invalid_bids, served) in [(7, Some(50)), (8, None)] { + let mut bid_values = vec![U256::from(50)]; + bid_values.resize(invalid_bids + 1, U256::from(60)); + let (_relay_state, relay) = start_stream_relay( + MockWsRelayState::new(chain, signer.clone()) + .with_bid_values(bid_values) + .with_invalid_last_bids(invalid_bids), + signer.public_key(), + ) + .await?; + + let validator = start_pbs(chain, vec![relay], 1_000).await?; + + let (code, res) = get_header_json(&validator).await?; + match served { + Some(value) => { + assert_eq!(code, StatusCode::OK, "{invalid_bids} invalid bids"); + assert_bid(&res.unwrap(), chain, &signer, U256::from(value)); + } + None => assert_eq!(code, StatusCode::NO_CONTENT, "{invalid_bids} invalid bids"), + } + } + Ok(()) +} + /// Frames PBS can't parse are skipped, not treated as the end of the stream: /// the updates after them still count. #[tokio::test] @@ -194,7 +230,7 @@ async fn test_get_header_ws_handshake_carries_request() -> Result<()> { start_stream_relay(MockWsRelayState::new(chain, signer.clone()), pubkey).await?; let validator = start_pbs(chain, vec![relay], timeout_ms).await?; - let sent_at = cb_common::utils::utcnow_ms(); + let sent_at = utcnow_ms(); let (code, _) = get_header_json(&validator).await?; assert_eq!(code, StatusCode::OK); @@ -203,8 +239,13 @@ async fn test_get_header_ws_handshake_carries_request() -> Result<()> { assert_eq!(request.parent_hash, B256::ZERO); assert!(request.validator_pubkey.starts_with("0x")); - // No timeout header from the caller, so PBS passes its own budget through - assert_eq!(request.timeout_ms, Some(timeout_ms)); + // No timeout header from the caller, so PBS asks for its own budget, less + // what the connect took + let relay_timeout_ms = request.timeout_ms.expect("missing timeout header"); + assert!( + (timeout_ms - 100..=timeout_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 + timeout_ms).contains(&start_time_ms)); @@ -247,6 +288,52 @@ async fn test_get_header_ws_returns_at_deadline() -> Result<()> { Ok(()) } +/// PBS asks the relay to end its stream half a TCP connect before PBS's own +/// deadline, so the last bid, one downlink trip later, still lands. +#[tokio::test] +async fn test_get_header_ws_last_bid_lands_before_deadline() -> Result<()> { + setup_test_env(); + let signer = random_secret(); + let chain = Chain::Hoodi; + let timeout_ms = 3_000; + + let relay_state = Arc::new( + MockWsRelayState::new(chain, signer.clone()) + .with_bid_values(vec![U256::from(10), U256::from(20)]) + .with_update_interval(Duration::from_millis(400)) + .ends_at_timeout(Duration::from_millis(200)), + ); + let (drain, drained) = oneshot::channel(); + let port = start_mock_ws_relay_with_full_backlog(relay_state.clone(), drained).await?; + let relay = generate_mock_stream_relay(port, signer.public_key())?; + let validator = start_pbs(chain, vec![relay], timeout_ms).await?; + + let requested_at_ms = utcnow_ms(); + let (res, ()) = tokio::join!(get_header_json(&validator), async { + // After PBS's first SYN, before its retransmit + tokio::time::sleep(Duration::from_millis(500)).await; + let _ = drain.send(()); + }); + let (code, res) = res?; + + assert_eq!(code, StatusCode::OK); + assert_bid(&res.unwrap(), chain, &signer, U256::from(20)); + + let request = relay_state.last_request().expect("relay saw no request"); + let connect_ms = request.accepted_at_ms - requested_at_ms; + assert!(connect_ms >= 500, "the connect was not delayed: {connect_ms}ms"); + + let relay_end_ms = request.start_time_ms.unwrap() + request.timeout_ms.unwrap(); + let early_ms = (requested_at_ms + timeout_ms) as i64 - relay_end_ms as i64; + let half_connect_ms = connect_ms as i64 / 2; + // PBS's delay before connecting skews both sides, hence the slack + assert!( + (half_connect_ms - 250..=half_connect_ms + 20).contains(&early_ms), + "relay asked to end {early_ms}ms before the deadline, connect took {connect_ms}ms" + ); + Ok(()) +} + /// A stream that never delivers a bid is a 204, same as an HTTP relay with no /// header for the slot. #[tokio::test] @@ -291,38 +378,10 @@ 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. +/// The HTTP request raced alongside the stream is the relay's normal +/// get_header, timing games included. #[tokio::test] -async fn test_get_header_ws_fallback_runs_timing_games() -> Result<()> { +async fn test_get_header_ws_race_http_runs_timing_games() -> Result<()> { setup_test_env(); let signer = random_secret(); let pubkey = signer.public_key(); @@ -343,7 +402,7 @@ async fn test_get_header_ws_fallback_runs_timing_games() -> Result<()> { 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"); + assert!(n_requests > 1, "HTTP skipped the timing games loop: {n_requests} requests"); Ok(()) } @@ -379,3 +438,76 @@ async fn test_get_header_ws_wins_auction_against_http() -> Result<()> { assert_eq!(stream_state.received_connections(), 1); Ok(()) } + +/// A handshake that outlasts the budget: the HTTP request raced alongside it +/// supplies the bid. +#[tokio::test] +async fn test_get_header_ws_race_handshake_timeout_http_bid_serves() -> Result<()> { + setup_test_env(); + let signer = random_secret(); + let chain = Chain::Hoodi; + let (http_state, ws_state, relay) = start_mock_dual_relay( + MockRelayState::new(chain, signer.clone()).with_bid_value(U256::from(42)), + MockWsRelayState::new(chain, signer.clone()).with_handshake_delay(Duration::from_secs(2)), + ) + .await?; + let validator = start_pbs(chain, vec![relay], 300).await?; + + let (code, res) = get_header_json(&validator).await?; + assert_eq!(code, StatusCode::OK); + assert_bid(&res.unwrap(), chain, &signer, U256::from(42)); + assert_eq!(http_state.received_get_header(), 1); + assert_eq!(ws_state.handshake_attempts(), 1); + assert_eq!(ws_state.received_connections(), 0, "the handshake never completed"); + Ok(()) +} + +/// A stream that breaks after a bid keeps that bid, here above the HTTP one. +#[tokio::test] +async fn test_get_header_ws_race_stream_break_keeps_held_bid() -> Result<()> { + setup_test_env(); + let signer = random_secret(); + let chain = Chain::Hoodi; + let (_, ws_state, relay) = start_mock_dual_relay( + MockRelayState::new(chain, signer.clone()).with_bid_value(U256::from(10)), + MockWsRelayState::new(chain, signer.clone()) + .with_bid_values(vec![U256::from(50)]) + .abort_after_bids(), + ) + .await?; + 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(50)); + assert_eq!(ws_state.received_connections(), 1); + Ok(()) +} + +/// The higher candidate wins, the stream's being its latest bid, not its +/// highest; a stream with no bid leaves the HTTP one. +#[tokio::test] +async fn test_get_header_ws_race_higher_bid_wins() -> Result<()> { + setup_test_env(); + let signer = random_secret(); + let chain = Chain::Hoodi; + + for (http_bid, stream_bids, served) in + [(10, vec![20, 50], 50), (60, vec![70, 50], 60), (42, vec![], 42)] + { + let (http_state, ws_state, relay) = start_mock_dual_relay( + MockRelayState::new(chain, signer.clone()).with_bid_value(U256::from(http_bid)), + MockWsRelayState::new(chain, signer.clone()) + .with_bid_values(stream_bids.into_iter().map(U256::from).collect()), + ) + .await?; + 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(served)); + assert_eq!(http_state.received_get_header(), 1); + assert_eq!(ws_state.received_connections(), 1); + } + Ok(()) +}