diff --git a/Cargo.lock b/Cargo.lock index d270840..33bfa39 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2802,7 +2802,7 @@ checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" [[package]] name = "vixy" -version = "0.1.3" +version = "0.1.4" dependencies = [ "axum", "clap", diff --git a/Cargo.toml b/Cargo.toml index 68e1898..0537dea 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "vixy" -version = "0.1.3" +version = "0.1.4" edition = "2024" description = "A Rust proxy that monitors Ethereum EL and CL nodes, tracks their health, and routes requests to healthy nodes" license = "MIT" diff --git a/src/config.rs b/src/config.rs index 3815c51..6baa1a3 100644 --- a/src/config.rs +++ b/src/config.rs @@ -37,6 +37,14 @@ pub struct Global { pub health_check_max_failures: u32, /// Maximum request body size in bytes (default: unlimited) pub max_body_size: usize, + /// Whether to mark an EL node unhealthy when its `newHeads` subscription stalls + /// (stops advancing) even though HTTP polling still succeeds. Detects upstreams + /// that serve a stale head over the subscription while otherwise looking healthy. + pub subscription_health_enabled: bool, + /// Time in milliseconds without a new head on vixy's own `newHeads` probe + /// subscription before the probe drops the connection and reconnects to + /// re-establish a fresh subscription. + pub subscription_stall_timeout_ms: u64, } /// Metrics configuration settings @@ -69,6 +77,8 @@ impl Default for Global { max_retries: 2, health_check_max_failures: 3, max_body_size: usize::MAX, + subscription_health_enabled: true, + subscription_stall_timeout_ms: 30000, } } } @@ -178,6 +188,16 @@ impl Config { fn validate(&self) -> Result<()> { self.el.validate().wrap_err("invalid EL configuration")?; + if self.global.subscription_health_enabled && self.global.subscription_stall_timeout_ms == 0 + { + return Err(ConfigError::ValidationFailed( + "subscription_stall_timeout_ms must be greater than 0 when \ + subscription_health_enabled is true" + .to_string(), + ) + .into()); + } + if self.cl.is_empty() { return Err(ConfigError::ValidationFailed( "at least one CL node is required".to_string(), diff --git a/src/health/el.rs b/src/health/el.rs index 0e05a13..4dd3bc4 100644 --- a/src/health/el.rs +++ b/src/health/el.rs @@ -98,7 +98,11 @@ pub fn calculate_el_health( // Calculate lag (how far behind the node is from chain head) node.lag = chain_head.saturating_sub(node.block_number); - // Determine if this check passed (check succeeded AND lag is within threshold) + // Determine if this check passed (check succeeded AND lag is within threshold). + // NOTE: newHeads-subscription freshness is deliberately NOT part of is_healthy — + // it is a separate, WebSocket-only signal (see state::ElNodeState::subscription_healthy + // and proxy::selection::select_el_ws_node) so a stalled subscription never diverts + // HTTP traffic or flips the failover flag. let check_passed = node.check_ok && node.lag <= max_lag; if check_passed { @@ -236,6 +240,9 @@ mod tests { is_healthy: false, lag: 0, consecutive_failures: 0, + sub_block_number: block_number, + sub_last_head_at: None, + subscription_healthy: true, } } @@ -378,6 +385,22 @@ mod tests { assert!(node.is_healthy, "Node should be healthy after recovery"); } + #[test] + fn test_subscription_staleness_does_not_affect_is_healthy() { + // A stalled newHeads subscription must NOT make is_healthy false: subscription + // freshness is a separate, WS-only signal. HTTP health depends only on the poll. + let mut node = make_el_node("test", 1000, true); + node.subscription_healthy = false; // subscription stale... + let chain_head = 1000; + + calculate_el_health(&mut node, chain_head, 5, 3); + + assert!( + node.is_healthy, + "is_healthy must reflect HTTP poll only, independent of subscription freshness" + ); + } + // ========================================================================= // update_el_chain_head tests // ========================================================================= diff --git a/src/health/mod.rs b/src/health/mod.rs index 70ccf10..b1fcf16 100644 --- a/src/health/mod.rs +++ b/src/health/mod.rs @@ -2,3 +2,4 @@ pub mod cl; pub mod el; +pub mod subscription; diff --git a/src/health/subscription.rs b/src/health/subscription.rs new file mode 100644 index 0000000..ab1ec6e --- /dev/null +++ b/src/health/subscription.rs @@ -0,0 +1,330 @@ +//! EL `newHeads` subscription liveness probe. +//! +//! Vixy maintains its own `newHeads` subscription to each EL upstream, independent of +//! the per-client WebSocket relay. It records the time of the latest head into shared +//! state so the monitor can detect when a node's subscription stalls (stops advancing) +//! even though HTTP polling still succeeds. +//! +//! This is the failure mode that otherwise goes undetected: an upstream's `newHeads` +//! subscription silently stops delivering while `eth_blockNumber` polling keeps +//! returning fresh heads, so the node looks healthy while every relayed client is +//! served a frozen head. +//! +//! Design notes: +//! - Detection is **time-since-last-head**, not a block-number comparison: a node is +//! subscription-stale when no head has arrived within `subscription_stall_timeout_ms`. +//! That timeout is the single, intuitive knob; it does not depend on `max_el_lag`. +//! - The resulting [`crate::state::ElNodeState::subscription_healthy`] flag is consumed +//! ONLY by WebSocket relay selection ([`crate::proxy::selection::select_el_ws_node`]). +//! It is never folded into `is_healthy`, so a stalled subscription does not divert +//! HTTP traffic or flip the failover flag. +//! - The probe's own stall timer keys on the last *head*, not on any frame, so WS-level +//! keepalive pings do not mask a head stall. + +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use futures_util::{SinkExt, StreamExt}; +use serde_json::Value; +use tokio_tungstenite::{connect_async, tungstenite::Message}; +use tracing::{debug, info, warn}; + +use crate::health::el::parse_hex_block_number; +use crate::state::AppState; + +/// Base delay between probe (re)connection attempts; grows exponentially per +/// consecutive *connection* failure up to [`PROBE_RECONNECT_BACKOFF_MAX`]. +const PROBE_RECONNECT_BACKOFF_BASE: Duration = Duration::from_secs(2); + +/// Maximum delay between probe (re)connection attempts. +const PROBE_RECONNECT_BACKOFF_MAX: Duration = Duration::from_secs(30); + +/// Timeout for establishing the probe WebSocket connection. +const PROBE_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); + +/// Whether a node's `newHeads` subscription is currently fresh. +/// +/// Returns `true` (do not gate) when the feature is disabled or the probe has never +/// delivered a head (`last_head_at` is `None`) — so a probe that has not yet connected +/// cannot, on its own, mark an otherwise-healthy node unhealthy. Once at least one head +/// has been seen, the subscription is fresh iff the most recent head arrived within +/// `stall_timeout`. +pub fn is_subscription_fresh( + last_head_at: Option, + stall_timeout: Duration, + enabled: bool, +) -> bool { + if !enabled { + return true; + } + match last_head_at { + None => true, + Some(t) => t.elapsed() <= stall_timeout, + } +} + +/// Spawn one `newHeads` probe task per configured EL node. +/// +/// Each probe maintains a persistent subscription and reconnects on stall/error. Returns +/// after fanning out the tasks (they run for the lifetime of the process). No-op when +/// subscription health gating is disabled in config. +pub async fn run_subscription_probes(state: Arc) { + if !state.subscription_health_enabled { + info!("EL subscription health gating disabled; not starting newHeads probes"); + return; + } + + // Snapshot the node list under a read lock, then release it. The node set is fixed + // at startup (AppState::new only ever pushes), so a one-time snapshot is complete. + let nodes: Vec<(String, String)> = { + let el_nodes = state.el_nodes.read().await; + el_nodes + .iter() + .map(|n| (n.name.clone(), n.ws_url.clone())) + .collect() + }; + + info!(count = nodes.len(), "Starting EL newHeads liveness probes"); + + for (name, ws_url) in nodes { + let state = state.clone(); + tokio::spawn(async move { + probe_node(state, name, ws_url).await; + }); + } +} + +/// Maintain a `newHeads` subscription to a single node, reconnecting forever on +/// stall/error/close. Connection failures back off exponentially (capped); a successful +/// connection resets the backoff. +async fn probe_node(state: Arc, name: String, ws_url: String) { + let stall_timeout = Duration::from_millis(state.subscription_stall_timeout_ms); + let mut consecutive_connect_failures: u32 = 0; + + loop { + match tokio::time::timeout(PROBE_CONNECT_TIMEOUT, connect_async(&ws_url)).await { + Ok(Ok((ws, _))) => { + consecutive_connect_failures = 0; + debug!(node = %name, "newHeads probe connected"); + if let Err(e) = run_probe_connection(&state, &name, ws, stall_timeout).await { + debug!(node = %name, error = %e, "newHeads probe connection ended"); + } + // Clean stall/close: reconnect promptly (base backoff). + tokio::time::sleep(PROBE_RECONNECT_BACKOFF_BASE).await; + } + Ok(Err(e)) => { + consecutive_connect_failures = consecutive_connect_failures.saturating_add(1); + let backoff = reconnect_backoff(consecutive_connect_failures); + warn!(node = %name, error = %e, backoff_secs = backoff.as_secs(), + "newHeads probe failed to connect, backing off"); + tokio::time::sleep(backoff).await; + } + Err(_) => { + consecutive_connect_failures = consecutive_connect_failures.saturating_add(1); + let backoff = reconnect_backoff(consecutive_connect_failures); + warn!(node = %name, timeout_secs = PROBE_CONNECT_TIMEOUT.as_secs(), + backoff_secs = backoff.as_secs(), + "newHeads probe connection timed out, backing off"); + tokio::time::sleep(backoff).await; + } + } + } +} + +/// Exponential backoff with a cap: base * 2^(failures-1), clamped to the max. +fn reconnect_backoff(consecutive_failures: u32) -> Duration { + let shift = consecutive_failures.saturating_sub(1).min(5); + PROBE_RECONNECT_BACKOFF_BASE + .saturating_mul(1u32 << shift) + .min(PROBE_RECONNECT_BACKOFF_MAX) +} + +/// Subscribe to `newHeads` on an established connection and record incoming heads until +/// the subscription stalls (no head within `stall_timeout`), errors, or closes. +/// +/// The stall timer keys on the last *head* only — non-head frames (WS keepalive pings, +/// the subscribe ack) do not reset it — so a head stall is detected even on a socket +/// that is otherwise kept alive. Returning drops the connection; the caller reconnects, +/// which re-establishes a fresh subscription (recovering the common case where an +/// upstream's per-subscription delivery stalls but the node is otherwise fine). +async fn run_probe_connection( + state: &Arc, + name: &str, + mut ws: S, + stall_timeout: Duration, +) -> Result<(), String> +where + S: futures_util::Stream> + + futures_util::Sink + + Unpin, +{ + let sub_req = + r#"{"jsonrpc":"2.0","id":1,"method":"eth_subscribe","params":["newHeads"]}"#.to_string(); + ws.send(Message::Text(sub_req.into())) + .await + .map_err(|e| format!("subscribe send failed: {e}"))?; + + debug!(node = %name, "newHeads probe subscribed"); + + // Deadline for the next head. Reset ONLY when a head arrives, never on other frames. + let mut head_deadline = tokio::time::Instant::now() + stall_timeout; + + loop { + tokio::select! { + msg = ws.next() => match msg { + Some(Ok(Message::Text(text))) => { + let text = text.as_str(); + if let Some(block) = parse_new_head_block(text) { + head_deadline = tokio::time::Instant::now() + stall_timeout; + record_head(state, name, block).await; + } else if let Some(err) = jsonrpc_error_message(text) { + // Surface a rejected subscription clearly instead of looping silently. + warn!(node = %name, error = %err, "newHeads subscription rejected by upstream"); + } + } + // Ping/pong/binary/close etc.: ignore and keep the head deadline running. + Some(Ok(_)) => {} + Some(Err(e)) => return Err(format!("ws error: {e}")), + None => return Ok(()), + }, + _ = tokio::time::sleep_until(head_deadline) => { + warn!( + node = %name, + timeout_secs = stall_timeout.as_secs(), + "newHeads probe received no head within stall timeout; reconnecting" + ); + return Ok(()); + } + } + } +} + +/// Extract the block number from a `newHeads` subscription notification, returning +/// `None` for any other message (subscribe ack, error, other subscription kinds). +fn parse_new_head_block(text: &str) -> Option { + let json: Value = serde_json::from_str(text).ok()?; + if json.get("method")?.as_str()? != "eth_subscription" { + return None; + } + let number = json.get("params")?.get("result")?.get("number")?.as_str()?; + parse_hex_block_number(number).ok() +} + +/// If `text` is a JSON-RPC error response, return its message (for logging). +fn jsonrpc_error_message(text: &str) -> Option { + let json: Value = serde_json::from_str(text).ok()?; + let error = json.get("error")?; + let message = error + .get("message") + .and_then(|m| m.as_str()) + .unwrap_or("unknown error"); + Some(message.to_string()) +} + +/// Record the latest subscription head for a node: store the block number (latest seen, +/// not a running max, so it stays accurate across reorgs/resyncs) and stamp the head +/// time used for freshness. +async fn record_head(state: &Arc, name: &str, block: u64) { + let mut el_nodes = state.el_nodes.write().await; + if let Some(node) = el_nodes.iter_mut().find(|n| n.name == name) { + node.sub_block_number = block; + node.sub_last_head_at = Some(Instant::now()); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_is_subscription_fresh_disabled_is_always_true() { + // Even a long-stale head is "fresh" when the feature is disabled. + let stale = Instant::now() - Duration::from_secs(3600); + assert!(is_subscription_fresh( + Some(stale), + Duration::from_secs(30), + false + )); + } + + #[test] + fn test_is_subscription_fresh_none_is_true() { + // Probe has never delivered a head → do not gate the node. + assert!(is_subscription_fresh(None, Duration::from_secs(30), true)); + } + + #[test] + fn test_is_subscription_fresh_recent_head_is_true() { + let recent = Instant::now() - Duration::from_secs(5); + assert!(is_subscription_fresh( + Some(recent), + Duration::from_secs(30), + true + )); + } + + #[test] + fn test_is_subscription_fresh_stale_head_is_false() { + // No head within the stall window → stale. + let stale = Instant::now() - Duration::from_secs(120); + assert!(!is_subscription_fresh( + Some(stale), + Duration::from_secs(30), + true + )); + } + + #[test] + fn test_reconnect_backoff_is_capped_and_grows() { + assert_eq!(reconnect_backoff(0), PROBE_RECONNECT_BACKOFF_BASE); // shift 0 + assert_eq!(reconnect_backoff(1), PROBE_RECONNECT_BACKOFF_BASE); // shift 0 + assert_eq!( + reconnect_backoff(2), + PROBE_RECONNECT_BACKOFF_BASE.saturating_mul(2) + ); + assert_eq!( + reconnect_backoff(3), + PROBE_RECONNECT_BACKOFF_BASE.saturating_mul(4) + ); + // Large failure counts are clamped to the max. + assert_eq!(reconnect_backoff(100), PROBE_RECONNECT_BACKOFF_MAX); + } + + #[test] + fn test_parse_new_head_block_valid() { + let frame = r#"{"jsonrpc":"2.0","method":"eth_subscription","params":{"subscription":"0xabc","result":{"number":"0x2cb6c9","hash":"0xdead"}}}"#; + assert_eq!(parse_new_head_block(frame), Some(0x2cb6c9)); + } + + #[test] + fn test_parse_new_head_block_ignores_subscribe_ack() { + let ack = r#"{"jsonrpc":"2.0","id":1,"result":"0xsubid"}"#; + assert_eq!(parse_new_head_block(ack), None); + } + + #[test] + fn test_parse_new_head_block_ignores_other_subscriptions() { + let logs = r#"{"jsonrpc":"2.0","method":"eth_subscription","params":{"subscription":"0xabc","result":{"address":"0x0","data":"0x"}}}"#; + assert_eq!(parse_new_head_block(logs), None); + } + + #[test] + fn test_parse_new_head_block_invalid_json() { + assert_eq!(parse_new_head_block("not json"), None); + } + + #[test] + fn test_jsonrpc_error_message_extracts_message() { + let err = r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32601,"message":"the method eth_subscribe does not exist"}}"#; + assert_eq!( + jsonrpc_error_message(err).as_deref(), + Some("the method eth_subscribe does not exist") + ); + } + + #[test] + fn test_jsonrpc_error_message_none_for_non_error() { + let ack = r#"{"jsonrpc":"2.0","id":1,"result":"0xsubid"}"#; + assert_eq!(jsonrpc_error_message(ack), None); + } +} diff --git a/src/main.rs b/src/main.rs index d9818b9..78bc3c3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -77,6 +77,13 @@ async fn main() -> eyre::Result<()> { info!(interval_ms = monitor_interval, "Health monitor started"); + // Start the EL newHeads subscription liveness probes. These detect upstreams whose + // newHeads subscription has silently stalled (serving a frozen head) even though + // HTTP polling still reports them healthy, and drive them subscription-unhealthy so + // the WS relay fails over (HTTP routing is unaffected). This fans out one background + // task per node and returns; no-op when gating is disabled in config. + vixy::health::subscription::run_subscription_probes(state.clone()).await; + // Initialize metrics if enabled (triggers lazy static initialization) if config.metrics.enabled { let _ = &*vixy::metrics::METRICS; diff --git a/src/metrics.rs b/src/metrics.rs index 37e24db..8357459 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -31,6 +31,10 @@ pub struct VixyMetrics { #[metric(rename = "el_node_healthy", labels = ["node", "tier"])] el_healthy: Gauge, + /// EL node newHeads subscription liveness (1=keeping up, 0=stalled) + #[metric(rename = "el_node_subscription_healthy", labels = ["node", "tier"])] + el_subscription_healthy: Gauge, + /// EL failover active status (1=active, 0=inactive) #[metric(rename = "el_failover_active")] el_failover_active: Gauge, @@ -154,6 +158,13 @@ impl VixyMetrics { .set(if healthy { 1u64 } else { 0u64 }); } + /// Set EL node newHeads subscription liveness (1 = keeping up, 0 = stalled) + pub fn set_el_subscription_healthy(node: &str, tier: &str, healthy: bool) { + METRICS + .el_subscription_healthy(node, tier) + .set(if healthy { 1u64 } else { 0u64 }); + } + /// Set EL failover active status pub fn set_el_failover_active(active: bool) { METRICS diff --git a/src/monitor.rs b/src/monitor.rs index fecefe2..3000b3e 100644 --- a/src/monitor.rs +++ b/src/monitor.rs @@ -9,7 +9,7 @@ use std::time::Duration; use futures_util::future; use tracing::{debug, info, warn}; -use crate::health::{cl, el}; +use crate::health::{cl, el, subscription}; use crate::metrics::VixyMetrics; use crate::state::AppState; @@ -97,8 +97,19 @@ pub async fn check_all_el_nodes(state: &Arc) -> bool { let mut el_nodes = state.el_nodes.write().await; let mut any_primary_healthy = false; let mut healthy_count = 0u64; + let sub_stall_timeout = Duration::from_millis(state.subscription_stall_timeout_ms); for node in el_nodes.iter_mut() { + // Recompute newHeads-subscription freshness (time since last head). This is a + // SEPARATE, WebSocket-only signal: it is consulted by select_el_ws_node, NOT + // folded into is_healthy, so a stalled subscription never affects HTTP routing + // or the failover flag. + node.subscription_healthy = subscription::is_subscription_fresh( + node.sub_last_head_at, + sub_stall_timeout, + state.subscription_health_enabled, + ); + el::calculate_el_health( node, chain_head, @@ -119,6 +130,11 @@ pub async fn check_all_el_nodes(state: &Arc) -> bool { VixyMetrics::set_el_block_number(&node.name, tier, node.block_number); VixyMetrics::set_el_lag(&node.name, tier, node.lag); VixyMetrics::set_el_healthy(&node.name, tier, node.is_healthy); + // Only emit subscription liveness once the probe is actually tracking this node, + // so a disabled feature or a not-yet-connected probe isn't reported as "fresh". + if state.subscription_health_enabled && node.sub_last_head_at.is_some() { + VixyMetrics::set_el_subscription_healthy(&node.name, tier, node.subscription_healthy); + } debug!( node = %node.name, @@ -126,6 +142,8 @@ pub async fn check_all_el_nodes(state: &Arc) -> bool { block_number = node.block_number, check_ok = node.check_ok, lag = node.lag, + sub_block_number = node.sub_block_number, + subscription_healthy = node.subscription_healthy, consecutive_failures = node.consecutive_failures, is_healthy = node.is_healthy, "EL node health calculated" diff --git a/src/proxy/http.rs b/src/proxy/http.rs index 1152c04..3afee72 100644 --- a/src/proxy/http.rs +++ b/src/proxy/http.rs @@ -275,6 +275,11 @@ pub struct ElNodeStatus { pub lag: u64, pub check_ok: bool, pub is_healthy: bool, + /// Latest block seen on vixy's own newHeads probe subscription to this node. + pub sub_block_number: u64, + /// Whether that subscription is currently fresh (WS-relay routing signal only; + /// does not affect `is_healthy`/HTTP routing). + pub subscription_healthy: bool, } /// CL node status for JSON response @@ -319,6 +324,8 @@ pub async fn status_handler(State(state): State>) -> Json Option<&E None } +/// Select an EL node for the WebSocket relay, where a stalled `newHeads` subscription +/// means clients would be served a frozen head. +/// +/// Prefers, in order: +/// 1. a healthy primary whose subscription is also fresh, +/// 2. any healthy node (primary or backup) whose subscription is fresh — a primary that +/// is HTTP-healthy but subscription-stale cannot serve WS, so a fresh backup is +/// preferred regardless of the HTTP failover flag, +/// 3. as a last resort, the plain health-based selection ([`select_el_node`]), so WS is +/// never *worse* off than HTTP during a total subscription outage. +pub fn select_el_ws_node(nodes: &[ElNodeState], failover_active: bool) -> Option<&ElNodeState> { + // 1. Healthy + fresh primary. + if let Some(n) = nodes + .iter() + .find(|n| n.is_primary && n.is_healthy && n.subscription_healthy) + { + return Some(n); + } + + // 2. Any healthy + fresh node (backups included, independent of failover_active). + if let Some(n) = nodes + .iter() + .find(|n| n.is_healthy && n.subscription_healthy) + { + return Some(n); + } + + // 3. Fall back to plain health-based selection (may be subscription-stale). + select_el_node(nodes, failover_active) +} + /// Select a healthy CL node /// /// Returns the first healthy CL node found. @@ -45,6 +76,9 @@ mod tests { is_healthy, lag: 0, consecutive_failures: 0, + sub_block_number: 1000, + sub_last_head_at: None, + subscription_healthy: true, } } @@ -111,6 +145,78 @@ mod tests { ); } + // ========================================================================= + // WebSocket EL node selection tests (subscription-aware) + // ========================================================================= + + #[test] + fn test_select_el_ws_prefers_fresh_primary() { + let mut nodes = vec![ + make_el_node("primary-1", true, true), + make_el_node("primary-2", true, true), + ]; + nodes[0].subscription_healthy = false; // stale sub + nodes[1].subscription_healthy = true; // fresh + + let selected = select_el_ws_node(&nodes, false); + + assert_eq!( + selected.unwrap().name, + "primary-2", + "WS selection should prefer a healthy primary whose subscription is fresh" + ); + } + + #[test] + fn test_select_el_ws_uses_fresh_backup_when_primary_sub_stale() { + // Primary is HTTP-healthy but subscription-stale; a fresh backup exists. Even + // with failover INACTIVE (HTTP primary is healthy), WS must avoid the stale + // primary and use the fresh backup — it cannot serve fresh heads. + let mut nodes = vec![ + make_el_node("primary-1", true, true), + make_el_node("backup-1", false, true), + ]; + nodes[0].subscription_healthy = false; + nodes[1].subscription_healthy = true; + + let selected = select_el_ws_node(&nodes, false); + + assert_eq!( + selected.unwrap().name, + "backup-1", + "WS selection should use a fresh backup when the only primary is subscription-stale" + ); + } + + #[test] + fn test_select_el_ws_falls_back_to_health_when_none_fresh() { + // No node has a fresh subscription → fall back to plain health-based selection + // (so WS is never worse off than HTTP during a total subscription outage). + let mut nodes = vec![ + make_el_node("primary-1", true, true), + make_el_node("backup-1", false, true), + ]; + nodes[0].subscription_healthy = false; + nodes[1].subscription_healthy = false; + + let selected = select_el_ws_node(&nodes, false); + + assert_eq!( + selected.unwrap().name, + "primary-1", + "WS selection should fall back to the healthy primary when no node is fresh" + ); + } + + #[test] + fn test_select_el_ws_none_when_no_healthy_node() { + let nodes = vec![make_el_node("primary-1", true, false)]; + assert!( + select_el_ws_node(&nodes, false).is_none(), + "WS selection should return None when no node is healthy" + ); + } + #[test] fn test_select_backup_when_no_primary_available() { let nodes = vec![ diff --git a/src/proxy/ws.rs b/src/proxy/ws.rs index 75df62a..1b8b74f 100644 --- a/src/proxy/ws.rs +++ b/src/proxy/ws.rs @@ -175,7 +175,7 @@ async fn is_node_healthy(state: &AppState, node_name: &str) -> bool { async fn select_healthy_node(state: &AppState) -> Option<(String, String)> { let failover_active = state.el_failover_active.load(Ordering::SeqCst); let el_nodes = state.el_nodes.read().await; - selection::select_el_node(&el_nodes, failover_active) + selection::select_el_ws_node(&el_nodes, failover_active) .map(|n| (n.name.clone(), n.ws_url.clone())) } @@ -247,8 +247,8 @@ pub async fn el_ws_handler(State(state): State>, ws: WebSocketUpgr let (ws_url, node_name) = { let el_nodes = state.el_nodes.read().await; - // Select a healthy node - match selection::select_el_node(&el_nodes, failover_active) { + // Select a healthy node whose newHeads subscription is also fresh + match selection::select_el_ws_node(&el_nodes, failover_active) { Some(n) => (n.ws_url.clone(), n.name.clone()), None => { warn!("No healthy EL node available for WebSocket"); @@ -943,6 +943,8 @@ mod tests { max_retries: 2, health_check_max_failures: 3, max_body_size: usize::MAX, + subscription_health_enabled: true, + subscription_stall_timeout_ms: 30000, http_client: reqwest::Client::new(), }) } @@ -958,6 +960,9 @@ mod tests { is_healthy, lag: 0, consecutive_failures: 0, + sub_block_number: 1000, + sub_last_head_at: None, + subscription_healthy: true, } } diff --git a/src/state.rs b/src/state.rs index 243e99c..fa4e656 100644 --- a/src/state.rs +++ b/src/state.rs @@ -5,6 +5,7 @@ use std::sync::Arc; use std::sync::atomic::AtomicBool; use std::sync::atomic::AtomicU64; +use std::time::Instant; use tokio::sync::RwLock; /// State for an EL (Execution Layer) node @@ -28,6 +29,19 @@ pub struct ElNodeState { pub lag: u64, /// Number of consecutive health check failures pub consecutive_failures: u32, + /// Latest block number observed on vixy's own `newHeads` probe subscription to this + /// node (informational; surfaced in `/status` and metrics). 0 until the first head. + pub sub_block_number: u64, + /// When the probe last received a `newHeads` notification from this node. `None` + /// until the first head arrives. Freshness is "a head within + /// `subscription_stall_timeout`"; a probe that has never delivered a head leaves + /// this `None` and therefore cannot, on its own, mark a node unhealthy. + pub sub_last_head_at: Option, + /// Whether the node's `newHeads` subscription is currently fresh. Computed each + /// monitor cycle from `sub_last_head_at`. Consulted ONLY by WebSocket relay + /// selection — deliberately NOT folded into `is_healthy`, so a stalled subscription + /// never diverts HTTP traffic or flips the failover flag. Defaults to true. + pub subscription_healthy: bool, } impl ElNodeState { @@ -43,6 +57,9 @@ impl ElNodeState { is_healthy: false, // Start unhealthy until health check passes lag: 0, consecutive_failures: 0, + sub_block_number: 0, + sub_last_head_at: None, + subscription_healthy: true, // Assume fresh until the probe proves otherwise } } } @@ -105,6 +122,10 @@ pub struct AppState { pub health_check_max_failures: u32, /// Maximum request body size in bytes pub max_body_size: usize, + /// Whether subscription-staleness health gating is enabled + pub subscription_health_enabled: bool, + /// Probe `newHeads` stall timeout in milliseconds (before the probe reconnects) + pub subscription_stall_timeout_ms: u64, /// Shared HTTP client for proxy requests (reuses connections) pub http_client: reqwest::Client, } @@ -147,6 +168,8 @@ impl AppState { max_retries: config.global.max_retries, health_check_max_failures: config.global.health_check_max_failures, max_body_size: config.global.max_body_size, + subscription_health_enabled: config.global.subscription_health_enabled, + subscription_stall_timeout_ms: config.global.subscription_stall_timeout_ms, http_client, } } diff --git a/tests/steps/el_health_steps.rs b/tests/steps/el_health_steps.rs index 76194eb..ba075ae 100644 --- a/tests/steps/el_health_steps.rs +++ b/tests/steps/el_health_steps.rs @@ -20,6 +20,9 @@ fn make_el_node(name: &str, block_number: u64, check_ok: bool) -> ElNodeState { is_healthy: false, lag: 0, consecutive_failures: 0, + sub_block_number: block_number, + sub_last_head_at: None, + subscription_healthy: true, } }