diff --git a/src/health/cl.rs b/src/health/cl.rs index 59dd7da..4229d5d 100644 --- a/src/health/cl.rs +++ b/src/health/cl.rs @@ -28,33 +28,27 @@ struct BeaconHeaderMessage { } /// Check if the CL node's health endpoint returns 200 -pub async fn check_cl_health(url: &str) -> Result { - // Use a timeout to prevent health checks from blocking indefinitely if the node is unresponsive - let client = reqwest::Client::builder() - .timeout(std::time::Duration::from_secs(5)) - .build() - .wrap_err("failed to build HTTP client")?; - +pub async fn check_cl_health(client: &reqwest::Client, url: &str) -> bool { let health_url = format!("{}/eth/v1/node/health", url.trim_end_matches('/')); - match client.get(&health_url).send().await { - Ok(response) => Ok(response.status().is_success()), - Err(_) => Ok(false), // Connection failure means unhealthy + match client + .get(&health_url) + .timeout(super::PROBE_TIMEOUT) + .send() + .await + { + Ok(response) => response.status().is_success(), + Err(_) => false, // Connection failure means unhealthy } } /// Get the current slot from the CL node's beacon headers endpoint -pub async fn check_cl_slot(url: &str) -> Result { - // Use a timeout to prevent health checks from blocking indefinitely if the node is unresponsive - let client = reqwest::Client::builder() - .timeout(std::time::Duration::from_secs(5)) - .build() - .wrap_err("failed to build HTTP client")?; - +pub async fn check_cl_slot(client: &reqwest::Client, url: &str) -> Result { let headers_url = format!("{}/eth/v1/beacon/headers/head", url.trim_end_matches('/')); let response = client .get(&headers_url) + .timeout(super::PROBE_TIMEOUT) .send() .await .wrap_err("failed to send request to CL node")?; @@ -73,14 +67,12 @@ pub async fn check_cl_slot(url: &str) -> Result { } /// Check both health and slot for a CL node -pub async fn check_cl_node(url: &str) -> Result<(bool, u64)> { - // Check health endpoint - let health_ok = check_cl_health(url).await?; +pub async fn check_cl_node(client: &reqwest::Client, url: &str) -> Result<(bool, u64)> { + // Run both probes concurrently: sequentially, an unreachable node costs two + // full timeouts per cycle, delaying the whole monitor pass. + let (health_ok, slot) = tokio::join!(check_cl_health(client, url), check_cl_slot(client, url)); - // Get current slot - let slot = check_cl_slot(url).await?; - - Ok((health_ok, slot)) + Ok((health_ok, slot?)) } /// Find the highest slot across all CL nodes (the chain head) @@ -137,9 +129,7 @@ mod tests { .mount(&mock_server) .await; - let result = check_cl_health(&mock_server.uri()) - .await - .expect("Should check health"); + let result = check_cl_health(&reqwest::Client::new(), &mock_server.uri()).await; assert!(result, "Should return true on 200 response"); } @@ -154,9 +144,7 @@ mod tests { .mount(&mock_server) .await; - let result = check_cl_health(&mock_server.uri()) - .await - .expect("Should check health"); + let result = check_cl_health(&reqwest::Client::new(), &mock_server.uri()).await; assert!(!result, "Should return false on 503 response"); } @@ -164,9 +152,7 @@ mod tests { #[tokio::test] async fn test_check_cl_health_returns_false_on_connection_failure() { // Use an invalid URL that will fail to connect - let result = check_cl_health("http://localhost:99999") - .await - .expect("Should handle connection failure"); + let result = check_cl_health(&reqwest::Client::new(), "http://localhost:99999").await; assert!(!result, "Should return false on connection failure"); } @@ -200,7 +186,7 @@ mod tests { .mount(&mock_server) .await; - let slot = check_cl_slot(&mock_server.uri()) + let slot = check_cl_slot(&reqwest::Client::new(), &mock_server.uri()) .await .expect("Should parse slot"); @@ -217,7 +203,7 @@ mod tests { .mount(&mock_server) .await; - let result = check_cl_slot(&mock_server.uri()).await; + let result = check_cl_slot(&reqwest::Client::new(), &mock_server.uri()).await; assert!(result.is_err(), "Should fail on invalid JSON"); } diff --git a/src/health/el.rs b/src/health/el.rs index 4dd3bc4..9bb9ea4 100644 --- a/src/health/el.rs +++ b/src/health/el.rs @@ -46,13 +46,7 @@ pub fn parse_hex_block_number(hex: &str) -> Result { } /// Check an EL node's current block number via JSON-RPC -pub async fn check_el_node(url: &str) -> Result { - // Use a timeout to prevent health checks from blocking indefinitely if the node is unresponsive - let client = reqwest::Client::builder() - .timeout(std::time::Duration::from_secs(5)) - .build() - .wrap_err("failed to build HTTP client")?; - +pub async fn check_el_node(client: &reqwest::Client, url: &str) -> Result { let request = JsonRpcRequest { jsonrpc: "2.0", method: "eth_blockNumber", @@ -62,6 +56,7 @@ pub async fn check_el_node(url: &str) -> Result { let response = client .post(url) + .timeout(super::PROBE_TIMEOUT) .json(&request) .send() .await @@ -191,7 +186,7 @@ mod tests { .mount(&mock_server) .await; - let block_number = check_el_node(&mock_server.uri()) + let block_number = check_el_node(&reqwest::Client::new(), &mock_server.uri()) .await .expect("Should get block number"); @@ -204,7 +199,7 @@ mod tests { // Don't mount any mock - request will fail - let result = check_el_node(&mock_server.uri()).await; + let result = check_el_node(&reqwest::Client::new(), &mock_server.uri()).await; assert!(result.is_err(), "Should fail on timeout/no response"); } @@ -221,7 +216,7 @@ mod tests { .mount(&mock_server) .await; - let result = check_el_node(&mock_server.uri()).await; + let result = check_el_node(&reqwest::Client::new(), &mock_server.uri()).await; assert!(result.is_err(), "Should fail on invalid hex in response"); } diff --git a/src/health/mod.rs b/src/health/mod.rs index b1fcf16..4614dc7 100644 --- a/src/health/mod.rs +++ b/src/health/mod.rs @@ -1,5 +1,13 @@ //! Health checking for EL and CL nodes +use std::time::Duration; + pub mod cl; pub mod el; pub mod subscription; + +/// Per-probe deadline, applied per request so a hung node can't stall a +/// monitor cycle. Probes run on the shared proxy client (`AppState::http_client`) +/// so they measure the same pooled connections that proxied traffic uses, +/// rather than paying a fresh DNS + TCP + TLS handshake per probe. +pub(crate) const PROBE_TIMEOUT: Duration = Duration::from_secs(5); diff --git a/src/monitor.rs b/src/monitor.rs index 3000b3e..050e202 100644 --- a/src/monitor.rs +++ b/src/monitor.rs @@ -16,21 +16,16 @@ use crate::state::AppState; /// Run a single health check cycle for all nodes /// /// This function checks all EL and CL nodes once and updates their state. +/// The EL and CL passes are independent, so they run concurrently. /// Returns true if at least one primary EL node is healthy. pub async fn run_health_check_cycle(state: &Arc) -> bool { - // Check all EL nodes - let any_primary_healthy = check_all_el_nodes(state).await; - - // Check all CL nodes - check_all_cl_nodes(state).await; - - // Update failover flag - update_failover_flag(state, any_primary_healthy); + let (any_primary_healthy, ()) = + tokio::join!(check_all_el_nodes(state), check_all_cl_nodes(state)); any_primary_healthy } -/// Check all EL nodes and update their state +/// Check all EL nodes, update their state, and refresh the failover flag /// /// Returns true if at least one primary EL node is healthy. pub async fn check_all_el_nodes(state: &Arc) -> bool { @@ -45,9 +40,11 @@ pub async fn check_all_el_nodes(state: &Arc) -> bool { .collect() }; - // Check all nodes concurrently without holding any lock + // Check all nodes concurrently without holding any lock. Probes ride the + // shared proxy client so they exercise the same pool as proxied traffic. + let client = &state.http_client; let check_results = future::join_all(node_checks.iter().map(|(name, url)| async move { - let result = el::check_el_node(url).await; + let result = el::check_el_node(client, url).await; (name.clone(), result) })) .await; @@ -153,6 +150,10 @@ pub async fn check_all_el_nodes(state: &Arc) -> bool { // Update healthy nodes count metric VixyMetrics::set_el_healthy_nodes(healthy_count); + // The flag is derived from the pass just completed; updating it here keeps + // node states and flag moving together as one EL health pass. + update_failover_flag(state, any_primary_healthy); + any_primary_healthy } @@ -170,8 +171,9 @@ pub async fn check_all_cl_nodes(state: &Arc) { }; // Check all nodes concurrently without holding any lock + let client = &state.http_client; let check_results = future::join_all(node_checks.iter().map(|(name, url)| async move { - let result = cl::check_cl_node(url).await; + let result = cl::check_cl_node(client, url).await; (name.clone(), result) })) .await; @@ -258,6 +260,10 @@ pub async fn check_all_cl_nodes(state: &Arc) { } /// Update the failover flag based on primary EL node availability +/// +/// The flag is observability-only (metrics, /status, transition logs). Node +/// selection derives failover from live node state and deliberately does not +/// consult this flag — see [`crate::proxy::selection::select_el_node`]. pub fn update_failover_flag(state: &Arc, any_primary_healthy: bool) { let was_failover = state.el_failover_active.load(Ordering::SeqCst); let is_failover = !any_primary_healthy; @@ -601,11 +607,8 @@ health_check_interval_ms = 100 // Initially failover should be inactive assert!(!state.el_failover_active.load(Ordering::SeqCst)); - // Run health check - primary will fail - let any_primary_healthy = check_all_el_nodes(&state).await; - - // Update failover flag - update_failover_flag(&state, any_primary_healthy); + // Run health check - primary will fail; the EL pass refreshes the flag itself + check_all_el_nodes(&state).await; // Failover should now be active since no primary is healthy assert!( @@ -614,6 +617,48 @@ health_check_interval_ms = 100 ); } + // ========================================================================= + // test_backup_usable_immediately_after_primary_marked_unhealthy + // ========================================================================= + + /// Regression test for the production 503s ("No healthy EL node available"): + /// the moment an EL pass marks the last primary unhealthy, a healthy backup + /// must be selectable and the failover flag must already be refreshed — no + /// separate cycle step may sit between node state and routing/observability. + #[tokio::test] + async fn test_backup_usable_immediately_after_primary_marked_unhealthy() { + let primary_mock = MockServer::start().await; + let backup_mock = MockServer::start().await; + + // Primary: no mock mounted → checks fail. Backup: healthy. + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "jsonrpc": "2.0", + "result": "0x3e8", + "id": 1 + }))) + .mount(&backup_mock) + .await; + + let config = create_config_with_backup(&[&primary_mock.uri()], &[&backup_mock.uri()], &[]); + let state = Arc::new(AppState::new(&config)); + + // Run enough EL passes to cross health_check_max_failures (default 3) + for _ in 0..state.health_check_max_failures { + check_all_el_nodes(&state).await; + } + + let el_nodes = state.el_nodes.read().await; + assert!(!el_nodes[0].is_healthy, "primary should be unhealthy"); + let selected = crate::proxy::selection::select_el_node(&el_nodes) + .expect("healthy backup must be selectable from live node state"); + assert_eq!(selected.name, "backup-0"); + assert!( + state.el_failover_active.load(Ordering::SeqCst), + "failover flag must be refreshed by the same EL pass" + ); + } + // ========================================================================= // test_monitor_clears_failover_when_primary_recovers // ========================================================================= @@ -649,11 +694,8 @@ health_check_interval_ms = 100 // Set failover as active (simulating previous failure) state.el_failover_active.store(true, Ordering::SeqCst); - // Run health check - primary is now healthy - let any_primary_healthy = check_all_el_nodes(&state).await; - - // Update failover flag - update_failover_flag(&state, any_primary_healthy); + // Run health check - primary is now healthy; the EL pass refreshes the flag itself + check_all_el_nodes(&state).await; // Failover should be cleared since primary is healthy assert!( diff --git a/src/proxy/http.rs b/src/proxy/http.rs index 3afee72..618b304 100644 --- a/src/proxy/http.rs +++ b/src/proxy/http.rs @@ -23,15 +23,12 @@ pub async fn el_proxy_handler( ) -> Response { let start = Instant::now(); - // Read the failover flag - let failover_active = state.el_failover_active.load(Ordering::SeqCst); - // Get a read lock on EL nodes and extract what we need let (target_url, node_name, tier) = { 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 (fails over to backups when no primary is healthy) + match selection::select_el_node(&el_nodes) { Some(n) => { let tier = if n.is_primary { "primary" } else { "backup" }; (n.http_url.clone(), n.name.clone(), tier) @@ -464,6 +461,60 @@ mod tests { assert_eq!(json["result"], "0x10d4f"); } + /// Regression: a healthy backup must serve traffic whenever no primary is + /// healthy, even while `el_failover_active` is false — routing must never + /// gate on the observability flag. + #[tokio::test] + async fn test_el_proxy_uses_backup_before_failover_flag_is_set() { + let backup_mock = MockServer::start().await; + + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "jsonrpc": "2.0", + "result": "0xbac", + "id": 1 + }))) + .mount(&backup_mock) + .await; + + let unhealthy_primary = make_el_node("geth-1", "http://localhost:1", false); + let mut backup = make_el_node("backup-1", &backup_mock.uri(), true); + backup.is_primary = false; + + let state = create_test_state(vec![unhealthy_primary, backup], vec![]); + assert!( + !state.el_failover_active.load(Ordering::SeqCst), + "precondition: failover flag not set" + ); + + let app = Router::new() + .route("/el", axum::routing::post(el_proxy_handler)) + .with_state(state); + + let request = Request::builder() + .method("POST") + .uri("/el") + .header("content-type", "application/json") + .body(Body::from( + r#"{"jsonrpc":"2.0","method":"eth_blockNumber","params":[],"id":1}"#, + )) + .unwrap(); + + let response = app.oneshot(request).await.unwrap(); + + assert_eq!( + response.status(), + StatusCode::OK, + "request must be served by the healthy backup, not refused with 503" + ); + + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let json: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(json["result"], "0xbac"); + } + #[tokio::test] async fn test_el_proxy_returns_503_no_healthy_nodes() { let el_nodes = vec![make_el_node("geth-1", "http://localhost:8545", false)]; // unhealthy diff --git a/src/proxy/selection.rs b/src/proxy/selection.rs index aa92317..8d44433 100644 --- a/src/proxy/selection.rs +++ b/src/proxy/selection.rs @@ -4,22 +4,14 @@ use crate::state::{ClNodeState, ElNodeState}; /// Select a healthy EL node, preferring primary nodes over backup /// -/// When failover_active is false, only primary nodes are considered. -/// When failover_active is true, both primary and backup nodes are considered. -pub fn select_el_node(nodes: &[ElNodeState], failover_active: bool) -> Option<&ElNodeState> { - // First try to find a healthy primary node - let primary = nodes.iter().find(|n| n.is_primary && n.is_healthy); - - if primary.is_some() { - return primary; - } - - // If no healthy primary and failover is active, try backup nodes - if failover_active { - return nodes.iter().find(|n| !n.is_primary && n.is_healthy); - } - - None +/// Failover derives from live node state: backups are eligible whenever no +/// healthy primary exists. The `el_failover_active` flag is a per-cycle +/// observability snapshot and deliberately does not gate routing. +pub fn select_el_node(nodes: &[ElNodeState]) -> Option<&ElNodeState> { + nodes + .iter() + .find(|n| n.is_primary && n.is_healthy) + .or_else(|| nodes.iter().find(|n| !n.is_primary && n.is_healthy)) } /// Select an EL node for the WebSocket relay, where a stalled `newHeads` subscription @@ -29,10 +21,10 @@ pub fn select_el_node(nodes: &[ElNodeState], failover_active: bool) -> Option<&E /// 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, +/// preferred, /// 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> { +pub fn select_el_ws_node(nodes: &[ElNodeState]) -> Option<&ElNodeState> { // 1. Healthy + fresh primary. if let Some(n) = nodes .iter() @@ -41,7 +33,7 @@ pub fn select_el_ws_node(nodes: &[ElNodeState], failover_active: bool) -> Option return Some(n); } - // 2. Any healthy + fresh node (backups included, independent of failover_active). + // 2. Any healthy + fresh node (backups included). if let Some(n) = nodes .iter() .find(|n| n.is_healthy && n.subscription_healthy) @@ -50,7 +42,7 @@ pub fn select_el_ws_node(nodes: &[ElNodeState], failover_active: bool) -> Option } // 3. Fall back to plain health-based selection (may be subscription-stale). - select_el_node(nodes, failover_active) + select_el_node(nodes) } /// Select a healthy CL node @@ -106,7 +98,7 @@ mod tests { make_el_node("geth-2", true, true), ]; - let selected = select_el_node(&nodes, false); + let selected = select_el_node(&nodes); assert!(selected.is_some(), "Should select a healthy node"); assert_eq!(selected.unwrap().name, "geth-1"); @@ -119,7 +111,7 @@ mod tests { make_el_node("geth-2", true, true), // healthy ]; - let selected = select_el_node(&nodes, false); + let selected = select_el_node(&nodes); assert!(selected.is_some(), "Should find a healthy node"); assert_eq!( @@ -136,12 +128,12 @@ mod tests { make_el_node("primary-1", true, true), // primary, healthy ]; - let selected = select_el_node(&nodes, true); // failover active + let selected = select_el_node(&nodes); assert!(selected.is_some()); assert!( selected.unwrap().is_primary, - "Should prefer primary over backup even when failover active" + "Should prefer primary over backup when both are healthy" ); } @@ -158,7 +150,7 @@ mod tests { nodes[0].subscription_healthy = false; // stale sub nodes[1].subscription_healthy = true; // fresh - let selected = select_el_ws_node(&nodes, false); + let selected = select_el_ws_node(&nodes); assert_eq!( selected.unwrap().name, @@ -179,7 +171,7 @@ mod tests { nodes[0].subscription_healthy = false; nodes[1].subscription_healthy = true; - let selected = select_el_ws_node(&nodes, false); + let selected = select_el_ws_node(&nodes); assert_eq!( selected.unwrap().name, @@ -199,7 +191,7 @@ mod tests { nodes[0].subscription_healthy = false; nodes[1].subscription_healthy = false; - let selected = select_el_ws_node(&nodes, false); + let selected = select_el_ws_node(&nodes); assert_eq!( selected.unwrap().name, @@ -212,7 +204,7 @@ mod tests { 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(), + select_el_ws_node(&nodes).is_none(), "WS selection should return None when no node is healthy" ); } @@ -224,20 +216,14 @@ mod tests { make_el_node("backup-1", false, true), // backup, healthy ]; - // Without failover, should return None (no healthy primary) - let without_failover = select_el_node(&nodes, false); - assert!( - without_failover.is_none(), - "Without failover, should not select backup" - ); - - // With failover, should select backup - let with_failover = select_el_node(&nodes, true); + // Failover is derived from live state: no healthy primary → backup is + // selected immediately, without waiting for any external flag. + let selected = select_el_node(&nodes); assert!( - with_failover.is_some(), - "With failover, should select backup" + selected.is_some(), + "Backup should be selected as soon as no primary is healthy" ); - assert_eq!(with_failover.unwrap().name, "backup-1"); + assert_eq!(selected.unwrap().name, "backup-1"); } #[test] @@ -247,7 +233,7 @@ mod tests { make_el_node("backup-1", false, false), // unhealthy ]; - let selected = select_el_node(&nodes, true); // even with failover + let selected = select_el_node(&nodes); assert!( selected.is_none(), @@ -259,7 +245,7 @@ mod tests { fn test_select_empty_list_returns_none() { let nodes: Vec = vec![]; - let selected = select_el_node(&nodes, true); + let selected = select_el_node(&nodes); assert!(selected.is_none(), "Empty list should return None"); } diff --git a/src/proxy/ws.rs b/src/proxy/ws.rs index 1b8b74f..880c5d9 100644 --- a/src/proxy/ws.rs +++ b/src/proxy/ws.rs @@ -9,7 +9,6 @@ use futures_util::{SinkExt, StreamExt}; use serde_json::Value; use std::collections::{HashMap, VecDeque}; use std::sync::Arc; -use std::sync::atomic::Ordering; use std::time::Duration; use tokio::sync::{Mutex, mpsc, oneshot}; use tokio_tungstenite::{connect_async, tungstenite::Message as TungsteniteMessage}; @@ -173,10 +172,8 @@ async fn is_node_healthy(state: &AppState, node_name: &str) -> bool { /// Select a new healthy node, returns (node_name, ws_url) 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_ws_node(&el_nodes, failover_active) - .map(|n| (n.name.clone(), n.ws_url.clone())) + selection::select_el_ws_node(&el_nodes).map(|n| (n.name.clone(), n.ws_url.clone())) } /// Health monitor task that watches for node health changes @@ -240,15 +237,13 @@ async fn health_monitor( /// Handle EL WebSocket upgrade requests (GET /el/ws) pub async fn el_ws_handler(State(state): State>, ws: WebSocketUpgrade) -> Response { - // Read the failover flag - let failover_active = state.el_failover_active.load(Ordering::SeqCst); - // Get a read lock on EL nodes and extract what we need let (ws_url, node_name) = { let el_nodes = state.el_nodes.read().await; // Select a healthy node whose newHeads subscription is also fresh - match selection::select_el_ws_node(&el_nodes, failover_active) { + // (fails over to backups when no primary qualifies) + match selection::select_el_ws_node(&el_nodes) { Some(n) => (n.ws_url.clone(), n.name.clone()), None => { warn!("No healthy EL node available for WebSocket"); @@ -981,8 +976,7 @@ mod tests { // Verify that node selection returns None when no healthy nodes let nodes = state.el_nodes.read().await; - let failover_active = state.el_failover_active.load(Ordering::SeqCst); - let selected = crate::proxy::selection::select_el_node(&nodes, failover_active); + let selected = crate::proxy::selection::select_el_node(&nodes); assert!(selected.is_none(), "Should not select unhealthy node"); } @@ -993,8 +987,7 @@ mod tests { // Verify that node selection returns the healthy node let nodes = state.el_nodes.read().await; - let failover_active = state.el_failover_active.load(Ordering::SeqCst); - let selected = crate::proxy::selection::select_el_node(&nodes, failover_active); + let selected = crate::proxy::selection::select_el_node(&nodes); assert!(selected.is_some(), "Should select healthy node"); assert_eq!(selected.unwrap().ws_url, "ws://localhost:8546"); } diff --git a/src/state.rs b/src/state.rs index fa4e656..f687004 100644 --- a/src/state.rs +++ b/src/state.rs @@ -108,7 +108,9 @@ pub struct AppState { pub el_chain_head: AtomicU64, /// Current CL chain head (highest slot seen) pub cl_chain_head: AtomicU64, - /// Whether we're in failover mode (using backup EL nodes) + /// Whether the last EL health pass found no healthy primary. Observability + /// only (metrics, /status, transition logs) — routing derives failover from + /// live node state, never from this flag. pub el_failover_active: AtomicBool, /// Maximum allowed EL lag in blocks pub max_el_lag: u64,