From ee1be53cf2581a54cb03f89954d7c005953043cd Mon Sep 17 00:00:00 2001 From: Jason Vranek Date: Thu, 10 Sep 2026 20:13:12 -0700 Subject: [PATCH] feat(pbs): count the ws bid stream under its own metrics endpoint --- crates/pbs/src/constants.rs | 13 +++++ crates/pbs/src/metrics.rs | 44 ++++++++++++++++ crates/pbs/src/mev_boost/get_header.rs | 6 ++- crates/pbs/src/mev_boost/get_header_ws.rs | 64 +++++++++++++++++------ docs/docs/get_started/running/metrics.md | 15 ++++++ 5 files changed, 125 insertions(+), 17 deletions(-) diff --git a/crates/pbs/src/constants.rs b/crates/pbs/src/constants.rs index 5976468db..3d5ee7258 100644 --- a/crates/pbs/src/constants.rs +++ b/crates/pbs/src/constants.rs @@ -1,8 +1,11 @@ +use reqwest::StatusCode; + pub const STATUS_ENDPOINT_TAG: &str = "status"; pub const REGISTER_VALIDATOR_ENDPOINT_TAG: &str = "register_validator"; pub const SUBMIT_BLINDED_BLOCK_ENDPOINT_TAG: &str = "submit_blinded_block"; pub const GET_HEADER_ENDPOINT_TAG: &str = "get_header"; pub const RELOAD_ENDPOINT_TAG: &str = "reload"; +pub const GET_HEADER_STREAM_ENDPOINT_TAG: &str = "get_header_stream"; /// For metrics recorded when a request times out pub const TIMEOUT_ERROR_CODE: u16 = 555; @@ -12,6 +15,16 @@ pub const TIMEOUT_ERROR_CODE_STR: &str = "555"; /// deadline expiring: refused, dns, tls, or a stream that broke mid-window pub const TRANSPORT_ERROR_CODE: u16 = 556; +pub const TIMEOUT_ERROR_STATUS: StatusCode = synthetic_status(TIMEOUT_ERROR_CODE); +pub const TRANSPORT_ERROR_STATUS: StatusCode = synthetic_status(TRANSPORT_ERROR_CODE); + +const fn synthetic_status(code: u16) -> StatusCode { + match StatusCode::from_u16(code) { + Ok(status) => status, + Err(_) => panic!("synthetic status codes are within the valid range"), + } +} + /// 20 MiB to cover edge cases for heavy blocks and also add a bit of slack for /// any Ethereum upgrades in the near future pub const MAX_SIZE_SUBMIT_BLOCK_RESPONSE: usize = 20 * 1024 * 1024; diff --git a/crates/pbs/src/metrics.rs b/crates/pbs/src/metrics.rs index 8ef049aa4..51846f946 100644 --- a/crates/pbs/src/metrics.rs +++ b/crates/pbs/src/metrics.rs @@ -52,6 +52,50 @@ lazy_static! { .unwrap(); + // THE WEBSOCKET BID STREAM + // Outcome and time-to-first-bid ride RELAY_STATUS_CODE / RELAY_LATENCY + // under `get_header_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"], + 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 + 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"], + 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 + 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"], + 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 + 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"], + PBS_METRICS_REGISTRY + ) + .unwrap(); + // TO BEACON NODE /// Status code returned to beacon node by endpoint pub static ref BEACON_NODE_STATUS: IntCounterVec = register_int_counter_vec_with_registry!( diff --git a/crates/pbs/src/mev_boost/get_header.rs b/crates/pbs/src/mev_boost/get_header.rs index 67fbad138..576ced51e 100644 --- a/crates/pbs/src/mev_boost/get_header.rs +++ b/crates/pbs/src/mev_boost/get_header.rs @@ -42,7 +42,10 @@ use crate::{ 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}, + metrics::{ + RELAY_HEADER_VALUE, RELAY_LAST_SLOT, RELAY_LATENCY, RELAY_STATUS_CODE, + RELAY_STREAM_FALLBACK, + }, state::{BuilderApiState, PbsState}, utils::check_gas_limit, }; @@ -263,6 +266,7 @@ async fn get_header_from_relay( 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)?; diff --git a/crates/pbs/src/mev_boost/get_header_ws.rs b/crates/pbs/src/mev_boost/get_header_ws.rs index afebec179..f0703f86e 100644 --- a/crates/pbs/src/mev_boost/get_header_ws.rs +++ b/crates/pbs/src/mev_boost/get_header_ws.rs @@ -42,10 +42,13 @@ use url::Url; use super::get_header::{RequestInfo, validate_get_header_response}; use crate::{ constants::{ - GET_HEADER_ENDPOINT_TAG, MAX_SIZE_GET_HEADER_RESPONSE, TIMEOUT_ERROR_CODE, - TRANSPORT_ERROR_CODE, + 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, }, - metrics::{RELAY_LATENCY, RELAY_STATUS_CODE}, mev_boost::get_header::decode_ssz_payload, }; @@ -82,7 +85,7 @@ pub(super) async fn get_header_ws( ) -> Result, PbsError> { let (status, res) = stream_header(request_info, relay, url, timeout_ms).await; RELAY_STATUS_CODE - .with_label_values(&[status.as_str(), GET_HEADER_ENDPOINT_TAG, &relay.id]) + .with_label_values(&[status.as_str(), GET_HEADER_STREAM_ENDPOINT_TAG, &relay.id]) .inc(); res } @@ -96,7 +99,7 @@ async fn stream_header( let deadline = Instant::now() + Duration::from_millis(timeout_ms); let request = match build_handshake_request(request_info, relay, &url, timeout_ms) { Ok(request) => request, - Err(err) => return (StatusCode::from_u16(TRANSPORT_ERROR_CODE).unwrap(), Err(err)), + Err(err) => return (TRANSPORT_ERROR_STATUS, Err(err)), }; let config = WebSocketConfig::default() @@ -114,13 +117,13 @@ async fn stream_header( Ok(Ok(connected)) => connected, Ok(Err(err)) => return connect_failed(&err), Err(_) => { - return ( - StatusCode::from_u16(TIMEOUT_ERROR_CODE).unwrap(), - Err(PbsError::WebSocketTimeout), - ); + 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); @@ -174,9 +177,16 @@ async fn stream_header( 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 Some((fork, bid_bytes)) = latest else { if let Some(err) = stream_error { - return (StatusCode::from_u16(TRANSPORT_ERROR_CODE).unwrap(), Err(err)); + return (TRANSPORT_ERROR_STATUS, Err(err)); } debug!(relay_id = relay.id.as_ref(), ?connect_latency, invalid_frames, "no header"); @@ -185,7 +195,7 @@ async fn stream_header( if let Some(first_bid_latency) = first_bid_latency { RELAY_LATENCY - .with_label_values(&[GET_HEADER_ENDPOINT_TAG, &relay.id]) + .with_label_values(&[GET_HEADER_STREAM_ENDPOINT_TAG, &relay.id]) .observe(first_bid_latency.as_secs_f64()); } @@ -224,10 +234,7 @@ async fn stream_header( /// the headers, so it can be partial or empty. fn connect_failed(err: &WsError) -> StreamOutcome { let WsError::Http(res) = err else { - return ( - StatusCode::from_u16(TRANSPORT_ERROR_CODE).unwrap(), - Err(PbsError::WebSocketConnect(err.to_string())), - ); + return (TRANSPORT_ERROR_STATUS, Err(PbsError::WebSocketConnect(err.to_string()))); }; let code = res.status(); @@ -238,6 +245,9 @@ fn connect_failed(err: &WsError) -> StreamOutcome { 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))) } @@ -374,10 +384,32 @@ mod tests { assert_eq!(status, StatusCode::NOT_FOUND); let (status, res) = connect_failed(&WsError::ConnectionClosed); - assert_eq!(status, StatusCode::from_u16(TRANSPORT_ERROR_CODE).unwrap()); + 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 = diff --git a/docs/docs/get_started/running/metrics.md b/docs/docs/get_started/running/metrics.md index 582001959..30ff9e606 100644 --- a/docs/docs/get_started/running/metrics.md +++ b/docs/docs/get_started/running/metrics.md @@ -92,3 +92,18 @@ datasources: Once Grafana is running, you can [import](https://grafana.com/docs/grafana/latest/dashboards/build-dashboards/import-dashboards/) the Commit-Boost dashboards from [here](https://github.com/Commit-Boost/commit-boost-client/tree/main/provisioning/grafana), making sure to select the correct `Prometheus` datasource. + +## Bid stream + +For a relay with `get_header = "stream"`, every series below is labelled by `relay_id`. The stream's own outcome and time-to-first-bid share the two general relay series under `endpoint="get_header_stream"`; the HTTP fallback keeps `endpoint="get_header"`, so the two divide slots served by the stream from slots served by the fallback. + +| 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, stream broke mid-window, or the handshake was answered with anything but `101`), 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. One at startup is the registration race; a steady rate means the relay is refusing the stream. The fallback's own results are under `endpoint="get_header"` | +| 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 | + +Two things the series do not tell apart. A bid that arrives but fails decoding or validation still counts as `200`, the same as over HTTP; the validation error is in the logs. And a request the beacon node abandons mid-window records no outcome at all.