websocket headers stream draft

.github/workflows/lint.yml

@@ -25,7 +25,7 @@ jobs:
       - name: Install Rust toolchain
         uses: dtolnay/rust-toolchain@master
         with:
-          toolchain: nightly-2025-09-18
+          toolchain: nightly-2026-07-26
           components: clippy, rustfmt
 
       - name: Setup just

crates/common/src/api/mod.rs

@@ -12,12 +12,18 @@ pub const PATH_GET_INCLUSION_LIST: &str = "/inclusion_list/{slot}/{parent_hash}/
 pub const PATH_DATA_ADJUSTMENTS: &str = "/adjustments";
 pub const PATH_MERGED_BLOCKS: &str = "/merged_blocks";
 
+/// How long the client gives us to answer, from `HEADER_START_TIME_UNIX_MS`
+pub const HEADER_TIMEOUT_MS: &str = "x-timeout-ms";
+/// When the client started the request, unix millis
+pub const HEADER_START_TIME_UNIX_MS: &str = "date-milliseconds";
+
 pub const PATH_PROPOSER_API: &str = "/eth/v1/builder";
 pub const PATH_PROPOSER_API_V2: &str = "/eth/v2/builder";
 
 pub const PATH_STATUS: &str = "/status";
 pub const PATH_REGISTER_VALIDATORS: &str = "/validators";
 pub const PATH_GET_HEADER: &str = "/header/{slot}/{parent_hash}/{pubkey}";
+pub const PATH_HEADER_STREAM: &str = "/header_stream/{slot}/{parent_hash}/{pubkey}";
 pub const PATH_GET_PAYLOAD: &str = "/blinded_blocks";
 
 pub const PATH_DATA_API: &str = "/relay/v1/data";

crates/common/src/api_provider.rs

@@ -4,7 +4,15 @@ use axum::http::HeaderMap;
 use helix_types::SignedValidatorRegistration;
 use tracing::warn;
 
-use crate::{PreferencesHeader, ValidatorPreferences, api::proposer_api::GetHeaderParams};
+use crate::{
+    PreferencesHeader, ValidatorPreferences,
+    api::{HEADER_TIMEOUT_MS, proposer_api::GetHeaderParams},
+};
+
+pub fn header_u64(headers: &HeaderMap, name: &str) -> Option<u64> {
+    let value = headers.get(name)?.to_str().ok()?.parse().ok()?;
+    (value > 0).then_some(value)
+}
 
 pub trait ApiProvider: Send + Sync + Clone + 'static {
     fn get_timing(
@@ -29,6 +37,7 @@ pub trait ApiProvider: Send + Sync + Clone + 'static {
 pub struct TimingResult {
     pub sleep_time: Option<Duration>,
     pub is_mev_boost: bool,
+    pub timeout_ms: Option<u64>,
 }
 
 #[derive(Clone)]
@@ -42,11 +51,15 @@ impl ApiProvider for DefaultApiProvider {
     fn get_timing(
         &self,
         _params: &GetHeaderParams,
-        _headers: &HeaderMap,
+        headers: &HeaderMap,
         _preferences: &ValidatorPreferences,
         _ms_into_slot: u64,
     ) -> Result<TimingResult, &'static str> {
-        Ok(TimingResult { sleep_time: None, is_mev_boost: false })
+        Ok(TimingResult {
+            sleep_time: None,
+            is_mev_boost: false,
+            timeout_ms: header_u64(headers, HEADER_TIMEOUT_MS),
+        })
     }
 
     fn get_preferences(

crates/common/src/config.rs

@@ -48,6 +48,9 @@ pub struct RelayConfig {
     /// Configuration for block merging parameters.
     #[serde(default)]
     pub block_merging_config: BlockMergingConfig,
+    /// Configuration for the websocket header stream.
+    #[serde(default)]
+    pub header_stream: HeaderStreamConfig,
     pub primev_config: Option<PrimevConfig>,
     pub discord_webhook_url: Option<Url>,
     pub alerts_config: Option<AlertsConfig>,
@@ -97,6 +100,7 @@ impl RelayConfig {
             router_config: Default::default(),
             target_get_payload_propagation_duration_ms: Default::default(),
             block_merging_config: Default::default(),
+            header_stream: Default::default(),
             primev_config: Default::default(),
             discord_webhook_url: Default::default(),
             alerts_config: Default::default(),
@@ -492,6 +496,7 @@ impl RouterConfig {
             Route::Status,
             Route::RegisterValidators,
             Route::GetHeader,
+            Route::HeaderStream,
             Route::GetPayload,
             Route::GetPayloadV2,
         ]);
@@ -541,8 +546,9 @@ impl RouterConfig {
             return Ok(true);
         }
 
-        let is_get_header_instance =
-            routes.contains(&Route::ProposerApi) || routes.contains(&Route::GetHeader);
+        let is_get_header_instance = routes.contains(&Route::ProposerApi) ||
+            routes.contains(&Route::GetHeader) ||
+            routes.contains(&Route::HeaderStream);
         let is_submission_instance =
             routes.contains(&Route::BuilderApi) || routes.contains(&Route::SubmitBlock);
 
@@ -602,6 +608,7 @@ pub enum Route {
     Status,
     RegisterValidators,
     GetHeader,
+    HeaderStream,
     GetPayload,
     GetPayloadV2,
     ProposerPayloadDelivered,
@@ -629,6 +636,7 @@ impl Route {
             Route::Status => format!("{PATH_PROPOSER_API}{PATH_STATUS}"),
             Route::RegisterValidators => format!("{PATH_PROPOSER_API}{PATH_REGISTER_VALIDATORS}"),
             Route::GetHeader => format!("{PATH_PROPOSER_API}{PATH_GET_HEADER}"),
+            Route::HeaderStream => format!("{PATH_PROPOSER_API}{PATH_HEADER_STREAM}"),
             Route::GetPayload => format!("{PATH_PROPOSER_API}{PATH_GET_PAYLOAD}"),
             Route::GetPayloadV2 => format!("{PATH_PROPOSER_API_V2}{PATH_GET_PAYLOAD}"),
             Route::ProposerPayloadDelivered => {
@@ -672,6 +680,25 @@ pub struct InclusionListConfig {
     pub relay_address: Address,
 }
 
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct HeaderStreamConfig {
+    /// How long to stream for, ending at the client's `X-Timeout-Ms` deadline.
+    #[serde(default = "default_u64::<300>")]
+    pub stream_for_ms: u64,
+    /// Interval between bid updates.
+    #[serde(default = "default_u64::<5>")]
+    pub interval_ms: u64,
+    /// Keys allowed to open a stream. Empty allows every connection.
+    #[serde(default)]
+    pub api_keys: HashSet<String>,
+}
+
+impl Default for HeaderStreamConfig {
+    fn default() -> Self {
+        Self { stream_for_ms: 300, interval_ms: 5, api_keys: HashSet::new() }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/relay/src/api/proposer/get_header.rs

@@ -1,6 +1,9 @@
 use std::{
-    sync::{Arc, atomic::Ordering},
-    time::Instant,
+    sync::{
+        Arc,
+        atomic::{AtomicBool, Ordering},
+    },
+    time::{Duration, Instant},
 };
 
 use axum::{Extension, extract::Path, http::HeaderMap, response::IntoResponse};
@@ -27,34 +30,27 @@ use crate::api::{
     router::Terminating,
 };
 
-impl<A: Api> ProposerApi<A> {
-    /// Retrieves the best bid header for the specified slot, parent hash, and public key.
-    ///
-    /// This function accepts a slot number, parent hash and public_key.
-    /// 1. Validates that the request's slot is not older than the head slot.
-    /// 2. Validates the request timestamp to ensure it's not too late.
-    /// 3. Fetches the best bid for the given parameters from the auctioneer.
-    ///
-    /// The function returns a JSON response containing the best bid if found.
-    ///
-    /// Implements this API: <https://ethereum.github.io/builder-specs/#/Builder/getHeader>
-    #[tracing::instrument(skip_all, err(level = tracing::Level::TRACE), fields(id =% extract_request_id(&headers), slot = params.slot, parent_hash =? params.parent_hash))]
-    pub async fn get_header(
-        Extension(proposer_api): Extension<Arc<ProposerApi<A>>>,
-        Extension(timings): Extension<RequestTimings>,
-        Extension(Terminating(terminating)): Extension<Terminating>,
-        headers: HeaderMap,
-        Path(params): Path<GetHeaderParams>,
-    ) -> Result<impl IntoResponse, ProposerApiError> {
-        trace!("starting call");
+pub(super) struct ValidatedHeaderRequest {
+    pub ms_into_slot: u64,
+    pub validation_complete_ns: u64,
+    pub user_agent: Option<String>,
+    pub is_mev_boost: bool,
+    pub sleep_time: Option<Duration>,
+    pub timeout_ms: Option<u64>,
+}
 
-        if terminating.load(Ordering::Relaxed) || proposer_api.local_cache.kill_switch_enabled() {
+impl<A: Api> ProposerApi<A> {
+    pub(super) fn validate_header_request(
+        &self,
+        params: &GetHeaderParams,
+        headers: &HeaderMap,
+        terminating: &AtomicBool,
+    ) -> Result<ValidatedHeaderRequest, ProposerApiError> {
+        if terminating.load(Ordering::Relaxed) || self.local_cache.kill_switch_enabled() {
             return Err(ProposerApiError::ServiceUnavailableError);
         }
 
-        let mut trace = GetHeaderTrace { receive: timings.on_receive_ns, ..Default::default() };
-
-        let (head_slot, duty) = proposer_api.curr_slot_info.slot_info();
+        let (head_slot, duty) = self.curr_slot_info.slot_info();
         let bid_slot = head_slot.as_u64() + 1;
 
         if params.slot != bid_slot {
@@ -68,18 +64,63 @@ impl<A: Api> ProposerApi<A> {
             return Err(ProposerApiError::ProposerNotRegistered);
         };
 
-        let ms_into_slot = validate_bid_request_time(&proposer_api.chain_info, &params)?;
-        trace.validation_complete = utcnow_ns();
+        let ms_into_slot = validate_bid_request_time(&self.chain_info, params)?;
+        let validation_complete_ns = utcnow_ns();
 
         trace!(ms_into_slot, "completed validation");
 
-        let user_agent = proposer_api.api_provider.get_metadata(&headers);
+        let user_agent = self.api_provider.get_metadata(headers);
 
-        let TimingResult { is_mev_boost, sleep_time } = proposer_api
+        let TimingResult { is_mev_boost, sleep_time, timeout_ms } = self
             .api_provider
-            .get_timing(&params, &headers, &duty.entry.preferences, ms_into_slot)
+            .get_timing(params, headers, &duty.entry.preferences, ms_into_slot)
             .map_err(ProposerApiError::InvalidGetHeader)?;
 
+        Ok(ValidatedHeaderRequest {
+            ms_into_slot,
+            validation_complete_ns,
+            user_agent,
+            is_mev_boost,
+            sleep_time,
+            timeout_ms,
+        })
+    }
+
+    /// Retrieves the best bid header for the specified slot, parent hash, and public key.
+    ///
+    /// This function accepts a slot number, parent hash and public_key.
+    /// 1. Validates that the request's slot is not older than the head slot.
+    /// 2. Validates the request timestamp to ensure it's not too late.
+    /// 3. Fetches the best bid for the given parameters from the auctioneer.
+    ///
+    /// The function returns a JSON response containing the best bid if found.
+    ///
+    /// Implements this API: <https://ethereum.github.io/builder-specs/#/Builder/getHeader>
+    #[tracing::instrument(skip_all, err(level = tracing::Level::TRACE), fields(id =% extract_request_id(&headers), slot = params.slot, parent_hash =? params.parent_hash))]
+    pub async fn get_header(
+        Extension(proposer_api): Extension<Arc<ProposerApi<A>>>,
+        Extension(timings): Extension<RequestTimings>,
+        Extension(Terminating(terminating)): Extension<Terminating>,
+        headers: HeaderMap,
+        Path(params): Path<GetHeaderParams>,
+    ) -> Result<impl IntoResponse, ProposerApiError> {
+        trace!("starting call");
+
+        let ValidatedHeaderRequest {
+            ms_into_slot,
+            validation_complete_ns,
+            user_agent,
+            is_mev_boost,
+            sleep_time,
+            ..
+        } = proposer_api.validate_header_request(&params, &headers, &terminating)?;
+
+        let mut trace = GetHeaderTrace {
+            receive: timings.on_receive_ns,
+            validation_complete: validation_complete_ns,
+            ..Default::default()
+        };
+
         let mut timing_guard = TimeoutGuard::default();
 
         if let Some(sleep_time) = sleep_time {
@@ -136,7 +177,7 @@ impl<A: Api> ProposerApi<A> {
             extra_data,
         );
 
-        let fork = proposer_api.chain_info.current_fork_name();
+        let fork = proposer_api.chain_info.fork_at_slot(params.slot.into());
         let payload_and_blobs = bid.payload_and_blobs();
         let bid_data = bid.bid_data_ref().to_owned();
         let bid = bid.into_builder_bid_slow();