fix(proxy): fail over to healthy EL backups from live node state

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<bool> {
-    // 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<u64> {
-    // 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<u64> {
     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<u64> {
 }
 
 /// 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,19 +144,15 @@ 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");
     }
 
     #[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");
     }
 

src/health/el.rs

@@ -46,13 +46,7 @@ pub fn parse_hex_block_number(hex: &str) -> Result<u64> {
 }
 
 /// Check an EL node's current block number via JSON-RPC
-pub async fn check_el_node(url: &str) -> Result<u64> {
-    // 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<u64> {
     let request = JsonRpcRequest {
         jsonrpc: "2.0",
         method: "eth_blockNumber",
@@ -62,6 +56,7 @@ pub async fn check_el_node(url: &str) -> Result<u64> {
 
     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");
     }
 

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);

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<AppState>) -> 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<AppState>) -> bool {
@@ -45,9 +40,11 @@ pub async fn check_all_el_nodes(state: &Arc<AppState>) -> 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<AppState>) -> 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<AppState>) {
     };
 
     // 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<AppState>) {
 }
 
 /// 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<AppState>, 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!(