From a7a64827526f9fb71f288f317b9491e35324b52d Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 1 Nov 2026 03:11:03 -0800 Subject: [PATCH] vector: follow a leader hint at once and skip one that failed this pass (#12295) * vector: follow a leader hint at once and skip one that failed this pass A follower naming a dead leader stalled master failover. Each pass broke off at the first hint, tried that leader first on the next pass, then got the same hint from the same follower, so the later configured masters were never tried. A pass now works like wdclient.tryAllMasters: it follows a hint at once, does not follow a hint to a master that already failed in that pass, and moves on to the rest. The hops are bounded by the master count, as the volume heartbeat bounds its hints, so two masters naming each other cannot spin. Co-Authored-By: Claude Opus 5.5 * vector: test a hint naming a failed master in its gRPC form A pass keys failed masters by host:http, so a configured m1:9333 that failed must also stop a hint of m1:9333.19333. Cover that equivalence. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK --------- Co-authored-by: Claude Opus 5.5 --- .../crates/server/src/master_client.rs | 108 +++++++++++++----- 1 file changed, 78 insertions(+), 30 deletions(-) diff --git a/seaweed-vector/crates/server/src/master_client.rs b/seaweed-vector/crates/server/src/master_client.rs index 31f14948c3..a1c357beda 100644 --- a/seaweed-vector/crates/server/src/master_client.rs +++ b/seaweed-vector/crates/server/src/master_client.rs @@ -2,6 +2,7 @@ //! connected: hold a KeepConnected stream open to the leader and the node //! shows up in ListClusterNodes(client_type="vector"). +use std::collections::HashSet; use std::time::Duration; use anyhow::{Context, Result}; @@ -67,37 +68,49 @@ fn http_addr(master: &str) -> String { /// Retry forever: try each master until a stream opens, hold it until it /// errors, then rediscover — same shape as wdclient.KeepConnectedToMaster. -/// A follower response's leader hint takes priority on the next pass. pub async fn keep_connected(masters: Vec, client_addr: String) { - let mut leader: Option = None; + let client_addr = client_addr.as_str(); loop { - for master in &seeds(&mut leader, &masters) { - match stream_to(master, &client_addr).await { - Ok(Some(l)) => { - leader = Some(l); - break; - } - Ok(None) => break, - Err(e) => tracing::warn!("master {master}: {e:#}"), - } - } + try_all_masters(&masters, |master| async move { + stream_to(&master, client_addr).await + }) + .await; sleep(Duration::from_secs(3)).await; } } -/// A hint is tried once, as wdclient.tryAllMasters keeps it for one cycle, so -/// a former leader that went away does not delay every later pass. -fn seeds(leader: &mut Option, masters: &[String]) -> Vec { - leader - .take() - .into_iter() - .chain(masters.iter().cloned()) - .collect() +/// One pass, as wdclient.tryAllMasters: a leader hint is followed at once, +/// and one that failed this pass is not followed again, so a follower naming +/// a dead leader cannot hide the other masters. +async fn try_all_masters(masters: &[String], mut connect: F) +where + F: FnMut(String) -> Fut, + Fut: Future>>, +{ + let mut failed = HashSet::new(); + for master in masters { + if failed.contains(&http_addr(master)) { + continue; + } + let mut target = master.clone(); + // Bounded, so two masters naming each other cannot spin. + for _ in 0..=masters.len() { + match connect(target.clone()).await { + Ok(Some(hint)) if !failed.contains(&http_addr(&hint)) => target = hint, + Ok(_) => break, + Err(e) => { + tracing::warn!("master {target}: {e:#}"); + failed.insert(http_addr(&target)); + break; + } + } + } + } } /// Bounds the TCP connect and then the master's first answer, the 5s wdclient /// gives a master call; a black-holed or silent master otherwise holds up the -/// other seeds. +/// other masters. const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); async fn stream_to(master: &str, client_addr: &str) -> Result> { @@ -153,15 +166,50 @@ async fn stream_to(master: &str, client_addr: &str) -> Result> { mod tests { use super::*; - #[test] - fn leader_hint_is_tried_once() { - let masters = vec!["m1:9333".to_string(), "m2:9333".to_string()]; - let mut leader = Some("m3:9333".to_string()); - assert_eq!( - seeds(&mut leader, &masters), - ["m3:9333", "m1:9333", "m2:9333"] - ); - assert_eq!(seeds(&mut leader, &masters), ["m1:9333", "m2:9333"]); + /// Runs one pass against scripted masters and returns the dial order. + async fn pass(masters: &[&str], answer: fn(&str) -> Result>) -> Vec { + let masters: Vec = masters.iter().map(|m| m.to_string()).collect(); + let mut tried = Vec::new(); + try_all_masters(&masters, |master| { + tried.push(master.clone()); + async move { answer(&master) } + }) + .await; + tried + } + + #[tokio::test] + async fn a_dead_leader_hint_does_not_hide_the_other_masters() { + let tried = pass(&["m1:9333", "m2:9333", "m3:9333"], |master| match master { + "m1:9333" | "m2:9333" => Ok(Some("dead:9333.19333".to_string())), + _ => anyhow::bail!("unreachable"), + }) + .await; + assert_eq!(tried, ["m1:9333", "dead:9333.19333", "m2:9333", "m3:9333"]); + } + + #[tokio::test] + async fn a_hint_naming_a_failed_master_in_grpc_form_is_skipped() { + let tried = pass(&["m1:9333", "m2:9333"], |master| match master { + "m1:9333" => anyhow::bail!("unreachable"), + _ => Ok(Some("m1:9333.19333".to_string())), + }) + .await; + assert_eq!(tried, ["m1:9333", "m2:9333"]); + } + + #[tokio::test] + async fn masters_naming_each_other_do_not_spin() { + let tried = pass(&["m1:9333", "m2:9333"], |master| { + let other = if master == "m1:9333" { + "m2:9333" + } else { + "m1:9333" + }; + Ok(Some(other.to_string())) + }) + .await; + assert_eq!(tried.len(), 6); } #[tokio::test]