feat(pbs): count the ws bid stream under its own metrics endpoint

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;

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!(

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, &params.parent_hash, &params.pubkey)?;

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<Option<GetHeaderResponse>, 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 =

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.