diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index d7901195b6..fc95c6964b 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/seaweed-volume/src/server/heartbeat.rs @@ -3,7 +3,7 @@ //! Implements the bidirectional streaming `SendHeartbeat` RPC to the master, //! matching Go's `server/volume_grpc_client_to_master.go`. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::path::Path; use std::sync::Arc; use std::sync::atomic::Ordering; @@ -13,7 +13,7 @@ use tokio::sync::broadcast; use tracing::{error, info, warn}; use super::grpc_client::{GrpcDialOptions, connect_channel, master_client}; -use super::volume_server::VolumeServerState; +use super::volume_server::{VolumeServerState, to_http_address}; use crate::pb::master_pb; use crate::pb::volume_server_pb; use crate::remote_storage::s3_tier::{S3TierBackend, S3TierConfig}; @@ -70,6 +70,7 @@ pub async fn run_heartbeat_with_state( let mut confirmed_leader_once = false; loop { + let mut tried_hints: HashSet = HashSet::new(); for master_addr in &config.master_addresses { if is_stopping(&state) { state.is_heartbeating.store(false, Ordering::Relaxed); @@ -82,112 +83,72 @@ pub async fn run_heartbeat_with_state( return; } - // If we have a leader redirect, sleep 3s then connect to the leader - // instead of iterating through the seed list - let target_addr = if let Some(ref leader) = new_leader { + // Named leaders are tried on top of the seed scan rather than in + // the next seed's place: a peer that keeps naming a dead leader + // would otherwise eat that seed's slot on every pass and starve + // it. A fresh name is followed right away — the pending seed may + // be the master the stream just lost — and a name already dialed + // this pass is not dialed again. A hint naming the upcoming seed + // is covered by the seed dial itself. Mirrors Go's heartbeat loop. + let mut hints = 0; + while let Some(leader) = new_leader.clone() { + let leader_http = to_http_address(&leader).into_owned(); + if leader_http == to_http_address(master_addr).as_ref() + || tried_hints.contains(&leader_http) + || hints >= config.master_addresses.len() + { + break; + } + tried_hints.insert(leader_http); + hints += 1; + // the new leader may actually is the same master + // need to wait a bit before adding itself tokio::time::sleep(Duration::from_secs(3)).await; - leader.clone() - } else { - master_addr.clone() - }; - - let grpc_addr = to_grpc_address(&target_addr); - info!("Connecting heartbeat to master {}", grpc_addr); - - // Determine what action to take after the heartbeat attempt. - // We convert the error to a string immediately so the non-Send - // Box is dropped before any .await point. - enum PostAction { - LeaderRedirect(String), - Done, - SleepDuplicate(Duration), - SleepPulse, + new_leader = heartbeat_once( + &config, + &state, + &leader, + pulse, + &mut shutdown_rx, + &mut confirmed_leader_once, + &mut duplicate_retry_count, + ) + .await; + if is_stopping(&state) { + break; + } } - let action = match do_heartbeat( + if is_stopping(&state) { + break; + } + if let Some(ref leader) = new_leader + && to_http_address(leader) == to_http_address(master_addr) + { + // the seed dial covers a hint that names the seed itself + new_leader = None; + } + tried_hints.insert(to_http_address(master_addr).into_owned()); + if let Some(named) = heartbeat_once( &config, &state, - &grpc_addr, - &target_addr, + master_addr, pulse, &mut shutdown_rx, &mut confirmed_leader_once, + &mut duplicate_retry_count, ) .await { - Ok(Some(leader)) => { - info!("Master leader changed to {}", leader); - PostAction::LeaderRedirect(leader) - } - Ok(None) => { - duplicate_retry_count = 0; - PostAction::Done - } - Err(e) => { - let err_msg = e.to_string(); - // Drop `e` (non-Send) before any .await - drop(e); - warn!("Heartbeat to {} error: {}", grpc_addr, err_msg); - - if err_msg.contains(DUPLICATE_UUID_RETRY_MESSAGE) { - if duplicate_retry_count >= MAX_DUPLICATE_UUID_RETRIES { - error!( - "Shut down Volume Server due to persistent duplicate volume directories after 3 retries" - ); - error!( - "Please check if another volume server is using the same directory" - ); - std::process::exit(1); - } - let retry_delay = duplicate_uuid_retry_delay(duplicate_retry_count); - duplicate_retry_count += 1; - warn!( - "Waiting {:?} before retrying due to duplicate UUID detection (attempt {}/3)...", - retry_delay, duplicate_retry_count - ); - PostAction::SleepDuplicate(retry_delay) - } else { - duplicate_retry_count = 0; - PostAction::SleepPulse - } - } - }; - - match action { - PostAction::LeaderRedirect(leader) => { - new_leader = Some(leader); - break; - } - PostAction::Done => { - new_leader = None; - } - PostAction::SleepDuplicate(delay) => { - new_leader = None; - tokio::time::sleep(delay).await; - } - PostAction::SleepPulse => { - new_leader = None; - // The lookup spends part of the retry delay rather than - // adding to it. - let retry_at = tokio::time::Instant::now() + pulse; - refresh_leader_from_peers(&config, &state, pulse).await; - let mut wait = retry_at.saturating_duration_since(tokio::time::Instant::now()); - if wait > Duration::from_secs(1) - && (!state.current_master().await.is_empty() || !confirmed_leader_once) - { - // A leader just named by the refresh usually accepts - // the next stream right away, and an unregistered - // server has nothing else to do, so reconnect - // promptly. Once the server was registered and no - // leader is known, keep the pulse backoff. - wait = Duration::from_secs(1); - } - tokio::time::sleep(wait).await; + // A hint the cap left untried stays pending: the seed's + // answer replaces it only once it was dialed. + let tried = new_leader + .as_ref() + .is_none_or(|p| tried_hints.contains(to_http_address(p).as_ref())); + if tried { + new_leader = Some(named); } } - - // If we connected to a leader (not seed list), break out after one attempt - // so we either reconnect to the new leader or fall back to seed list - if new_leader.is_some() { + if is_stopping(&state) { break; } } @@ -198,6 +159,110 @@ pub async fn run_heartbeat_with_state( } } +/// One heartbeat attempt on `target_addr`; returns the leader the stream or +/// the peer refresh named — `None` when neither did. Mirrors Go's +/// `heartbeatOnce`. +async fn heartbeat_once( + config: &HeartbeatConfig, + state: &Arc, + target_addr: &str, + pulse: Duration, + shutdown_rx: &mut broadcast::Receiver<()>, + confirmed_leader_once: &mut bool, + duplicate_retry_count: &mut u32, +) -> Option { + let grpc_addr = to_grpc_address(target_addr); + info!("Connecting heartbeat to master {}", grpc_addr); + + // Determine what action to take after the heartbeat attempt. + // We convert the error to a string immediately so the non-Send + // Box is dropped before any .await point. + enum PostAction { + SleepDuplicate(Duration), + SleepPulse, + } + let action = match do_heartbeat( + config, + state, + &grpc_addr, + target_addr, + pulse, + shutdown_rx, + confirmed_leader_once, + ) + .await + { + Ok(Some(leader)) => { + info!("Master leader changed to {}", leader); + return Some(leader); + } + Ok(None) => { + *duplicate_retry_count = 0; + return None; + } + Err(e) => { + let err_msg = e.to_string(); + // Drop `e` (non-Send) before any .await + drop(e); + warn!("Heartbeat to {} error: {}", grpc_addr, err_msg); + + if err_msg.contains(DUPLICATE_UUID_RETRY_MESSAGE) { + if *duplicate_retry_count >= MAX_DUPLICATE_UUID_RETRIES { + error!( + "Shut down Volume Server due to persistent duplicate volume directories after 3 retries" + ); + error!("Please check if another volume server is using the same directory"); + std::process::exit(1); + } + let retry_delay = duplicate_uuid_retry_delay(*duplicate_retry_count); + *duplicate_retry_count += 1; + warn!( + "Waiting {:?} before retrying due to duplicate UUID detection (attempt {}/3)...", + retry_delay, duplicate_retry_count + ); + PostAction::SleepDuplicate(retry_delay) + } else { + *duplicate_retry_count = 0; + PostAction::SleepPulse + } + } + }; + + match action { + PostAction::SleepDuplicate(delay) => { + tokio::time::sleep(delay).await; + } + PostAction::SleepPulse => { + // The lookup spends part of the retry delay rather than adding + // to it. + let retry_at = tokio::time::Instant::now() + pulse; + let leader = refresh_leader_from_peers(config, state, target_addr, pulse).await; + let mut wait = retry_at.saturating_duration_since(tokio::time::Instant::now()); + if wait > Duration::from_secs(1) + && (!state.current_master().await.is_empty() || !*confirmed_leader_once) + { + // A leader just named by the refresh usually accepts the + // next stream right away, and an unregistered server has + // nothing else to do, so reconnect promptly. Once the + // server was registered and no leader is known, keep the + // pulse backoff. + wait = Duration::from_secs(1); + } + tokio::time::sleep(wait).await; + if let Some(leader) = leader + && to_http_address(&leader) != to_http_address(target_addr) + { + // Dial the leader the other masters just named rather than + // the next seed, which can be the master this stream just + // lost: an unreachable master holds each dial for the + // connect timeout before the loop moves on. + return Some(leader); + } + } + } + None +} + /// Convert a master address to a gRPC `host:port` target. /// /// Mirrors Go's `pb.ServerToGrpcAddress()`: @@ -274,22 +339,47 @@ pub async fn try_get_master_configuration( /// to the master the heartbeat just lost while the loop retries. Without an /// answer inside `timeout` the published master is cleared: the lost master /// is dead or no longer leader, and a follower answers lookups from its own -/// cache rather than refusing them. Mirrors Go's refreshLeaderFromPeers. +/// cache rather than refusing them, and returns it so the loop can dial it +/// next. The lost master is asked last: an unreachable one holds its lookup +/// open long enough to starve the peers that know the new leader. Mirrors +/// Go's refreshLeaderFromPeers. async fn refresh_leader_from_peers( config: &HeartbeatConfig, state: &Arc, + lost: &str, timeout: Duration, -) { - let lookup = lookup_leader_from_peers( - &config.master_addresses, - state.outgoing_grpc_tls.as_ref(), - ); +) -> Option { + let ordered = masters_asking_last(&config.master_addresses, lost); + let lookup = lookup_leader_from_peers(&ordered, state.outgoing_grpc_tls.as_ref()); match tokio::time::timeout(timeout, lookup).await { - Ok(Some(leader)) => publish_current_master(state, &leader).await, - _ => state.current_master_url.write().await.clear(), + Ok(Some(leader)) => { + publish_current_master(state, &leader).await; + Some(leader) + } + _ => { + state.current_master_url.write().await.clear(); + None + } } } +/// Return masters with every entry equal to `lost` (same HTTP address) moved +/// to the end, keeping the order of the rest. Mirrors Go's mastersAskingLast. +fn masters_asking_last(masters: &[String], lost: &str) -> Vec { + let lost_http = to_http_address(lost); + let mut ordered = Vec::with_capacity(masters.len()); + let mut tail = Vec::new(); + for m in masters { + if !lost.is_empty() && to_http_address(m) == lost_http { + tail.push(m.clone()); + } else { + ordered.push(m.clone()); + } + } + ordered.extend(tail); + ordered +} + /// Ask each seed master for the raft leader and return the first answer, or /// None when none replies with one. Mirrors Go's /// `operation.LookupRaftLeaderMaster`. @@ -1619,6 +1709,26 @@ mod tests { assert_eq!(to_grpc_address("host:not-a-port"), "host:not-a-port"); } + // Mirrors Go's TestMastersAskingLast: the just-lost master is asked last + // so an unreachable one cannot spend the whole refresh budget. + #[test] + fn test_masters_asking_last() { + let seeds = vec![ + "a:9333".to_string(), + "b:9333".to_string(), + "c:9333".to_string(), + ]; + assert_eq!( + masters_asking_last(&seeds, "a:9333.19333"), + vec![ + "b:9333".to_string(), + "c:9333".to_string(), + "a:9333".to_string() + ] + ); + assert_eq!(masters_asking_last(&seeds[..2], ""), seeds[..2]); + } + #[test] fn test_build_heartbeat_includes_store_identity_and_disk_metadata() { let temp_dir = tempfile::tempdir().unwrap(); diff --git a/weed/server/volume_grpc_client_to_master.go b/weed/server/volume_grpc_client_to_master.go index 5237cf2b21..bb48e6f14c 100644 --- a/weed/server/volume_grpc_client_to_master.go +++ b/weed/server/volume_grpc_client_to_master.go @@ -35,10 +35,14 @@ func (vs *VolumeServer) GetMaster(ctx context.Context) pb.ServerAddress { // the heartbeat loop owns that field and reconnects when the stream reports a // different leader than the connected peer. func (vs *VolumeServer) lookupRaftLeaderMaster(ctx context.Context) (pb.ServerAddress, error) { + return vs.lookupRaftLeaderMasterAmong(ctx, vs.SeedMasterNodes) +} + +func (vs *VolumeServer) lookupRaftLeaderMasterAmong(ctx context.Context, masters []pb.ServerAddress) (pb.ServerAddress, error) { if ctx == nil { ctx = context.Background() } - leader, err := operation.LookupRaftLeaderMaster(ctx, vs.SeedMasterNodes, vs.grpcDialOption) + leader, err := operation.LookupRaftLeaderMaster(ctx, masters, vs.grpcDialOption) if err != nil { if isContextDoneErr(err) { glog.V(1).Infof("volume server %s:%d: raft leader lookup: %v", vs.store.Ip, vs.store.Port, err) @@ -56,18 +60,40 @@ func (vs *VolumeServer) lookupRaftLeaderMaster(ctx context.Context) (pb.ServerAd // refreshLeaderFromPeers publishes the leader the configured masters report, // so lookups stop going to a master the heartbeat just lost while the loop -// retries. Without an answer within timeout no leader is published: the lost -// master is dead or no longer leader, and a follower answers lookups from its -// own cache rather than refusing them, so failing fast is the safer answer. -func (vs *VolumeServer) refreshLeaderFromPeers(timeout time.Duration) { +// retries, and returns it so the loop can dial it next. Without an answer +// within timeout no leader is published: the lost master is dead or no longer +// leader, and a follower answers lookups from its own cache rather than +// refusing them, so failing fast is the safer answer. +// +// The lost master is asked last. Each peer gets up to 5s, so when the lost +// master is unreachable rather than refusing connections (a partition or a +// powered-off host) asking it first spends the whole timeout and the live +// peers that know the new leader are never asked. +func (vs *VolumeServer) refreshLeaderFromPeers(lost pb.ServerAddress, timeout time.Duration) pb.ServerAddress { ctx, cancel := context.WithTimeout(context.Background(), timeout) defer cancel() - leader, err := vs.lookupRaftLeaderMaster(ctx) + leader, err := vs.lookupRaftLeaderMasterAmong(ctx, mastersAskingLast(vs.SeedMasterNodes, lost)) if err != nil { vs.setCurrentMaster("") - return + return "" } vs.setCurrentMaster(leader) + return leader +} + +// mastersAskingLast returns masters with every entry equal to last moved to +// the end, keeping the order of the rest. +func mastersAskingLast(masters []pb.ServerAddress, last pb.ServerAddress) []pb.ServerAddress { + ordered := make([]pb.ServerAddress, 0, len(masters)) + var tail []pb.ServerAddress + for _, m := range masters { + if last != "" && m.Equals(last) { + tail = append(tail, m) + continue + } + ordered = append(ordered, m) + } + return append(ordered, tail...) } func isContextDoneErr(err error) bool { @@ -135,51 +161,42 @@ func (vs *VolumeServer) heartbeat() { grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.volume") - var err error var newLeader pb.ServerAddress duplicateRetryCount := 0 for vs.isHeartbeating { + triedHints := make(map[string]bool) for _, master := range vs.SeedMasterNodes { - if newLeader != "" { + // Named leaders are tried on top of the seed scan rather than in + // the next seed's place: a peer that keeps naming a dead leader + // would otherwise eat that seed's slot on every pass and starve + // it. A fresh name is followed right away — the pending seed may + // be the master the stream just lost — and a name already dialed + // this pass is not dialed again. + for hints := 0; newLeader != "" && !newLeader.Equals(master) && !triedHints[newLeader.ToHttpAddress()] && hints < len(vs.SeedMasterNodes); hints++ { // the new leader may actually is the same master // need to wait a bit before adding itself time.Sleep(3 * time.Second) - master = newLeader - } - newLeader, err = vs.doHeartbeatWithRetry(master, grpcDialOption, vs.pulsePeriod, duplicateRetryCount) - if err != nil { - glog.V(0).Infof("heartbeat to %s error: %v", master, err) - - // Check if this is a duplicate UUID retry error - if strings.Contains(err.Error(), "duplicate UUIDs detected, retrying connection") { - duplicateRetryCount++ - retryDelay := time.Duration(1<<(duplicateRetryCount-1)) * 2 * time.Second // exponential backoff: 2s, 4s, 8s - glog.V(0).Infof("Waiting %v before retrying due to duplicate UUID detection...", retryDelay) - time.Sleep(retryDelay) - } else { - // Regular error, reset duplicate retry count - duplicateRetryCount = 0 - // The lookup spends part of the retry delay rather than - // adding to it. - retryAt := time.Now().Add(vs.pulsePeriod) - vs.refreshLeaderFromPeers(vs.pulsePeriod) - wait := time.Until(retryAt) - if wait > time.Second && (vs.getCurrentMaster() != "" || !vs.confirmedLeaderOnce.Load()) { - // A leader just named by the refresh usually accepts the - // next stream right away, and an unregistered server has - // nothing else to do, so reconnect promptly. Once the - // server was registered and no leader is known, keep the - // pulse backoff. - wait = time.Second - } - time.Sleep(wait) + triedHints[newLeader.ToHttpAddress()] = true + newLeader, duplicateRetryCount = vs.heartbeatOnce(newLeader, grpcDialOption, duplicateRetryCount) + if !vs.isHeartbeating { + break } - - stats.VolumeServerMasterDisconnections.WithLabelValues(master.String()).Inc() + } + if !vs.isHeartbeating { + break + } + if newLeader.Equals(master) { + // the seed dial covers a hint that names the seed itself newLeader = "" - } else { - // Successful connection, reset retry count - duplicateRetryCount = 0 + } + triedHints[master.ToHttpAddress()] = true + named, retryCount := vs.heartbeatOnce(master, grpcDialOption, duplicateRetryCount) + duplicateRetryCount = retryCount + // A hint the cap left untried stays pending: the seed's answer + // replaces it only once it was dialed, so a stale name cannot + // keep a live leader from ever being tried. + if named != "" && (newLeader == "" || triedHints[newLeader.ToHttpAddress()]) { + newLeader = named } if !vs.isHeartbeating { break @@ -188,6 +205,56 @@ func (vs *VolumeServer) heartbeat() { } } +// heartbeatOnce makes one heartbeat attempt on target and returns the leader +// the stream or the peer refresh named — "" when neither did — with the +// running duplicate-UUID retry count. +func (vs *VolumeServer) heartbeatOnce(target pb.ServerAddress, grpcDialOption grpc.DialOption, duplicateRetryCount int) (named pb.ServerAddress, retryCount int) { + newLeader, err := vs.doHeartbeatWithRetry(target, grpcDialOption, vs.pulsePeriod, duplicateRetryCount) + if err == nil { + // Successful connection, reset retry count + return newLeader, 0 + } + + glog.V(0).Infof("heartbeat to %s error: %v", target, err) + + var leader pb.ServerAddress + retryCount = duplicateRetryCount + // Check if this is a duplicate UUID retry error + if strings.Contains(err.Error(), "duplicate UUIDs detected, retrying connection") { + retryCount++ + retryDelay := time.Duration(1<<(retryCount-1)) * 2 * time.Second // exponential backoff: 2s, 4s, 8s + glog.V(0).Infof("Waiting %v before retrying due to duplicate UUID detection...", retryDelay) + time.Sleep(retryDelay) + } else { + // Regular error, reset duplicate retry count + retryCount = 0 + // The lookup spends part of the retry delay rather than + // adding to it. + retryAt := time.Now().Add(vs.pulsePeriod) + leader = vs.refreshLeaderFromPeers(target, vs.pulsePeriod) + wait := time.Until(retryAt) + if wait > time.Second && (vs.getCurrentMaster() != "" || !vs.confirmedLeaderOnce.Load()) { + // A leader just named by the refresh usually accepts the + // next stream right away, and an unregistered server has + // nothing else to do, so reconnect promptly. Once the + // server was registered and no leader is known, keep the + // pulse backoff. + wait = time.Second + } + time.Sleep(wait) + } + + stats.VolumeServerMasterDisconnections.WithLabelValues(target.String()).Inc() + if leader != "" && !leader.Equals(target) { + // Dial the leader the other masters just named rather than + // the next seed, which can be the master this stream just + // lost: an unreachable master holds each dial for the gRPC + // connect timeout before the loop moves on. + return leader, retryCount + } + return "", retryCount +} + func (vs *VolumeServer) StopHeartbeat() (isAlreadyStopping bool) { if !vs.isHeartbeating { return true diff --git a/weed/server/volume_grpc_client_to_master_test.go b/weed/server/volume_grpc_client_to_master_test.go index b235e66ca5..e1c55e6d68 100644 --- a/weed/server/volume_grpc_client_to_master_test.go +++ b/weed/server/volume_grpc_client_to_master_test.go @@ -63,8 +63,15 @@ func startFakeMasterServerForLeaderLookup(t *testing.T, srv master_pb.SeaweedSer go func() { _ = grpcServer.Serve(lis) }() t.Cleanup(grpcServer.GracefulStop) - _, port, _ := net.SplitHostPort(lis.Addr().String()) - return pb.ServerAddress(fmt.Sprintf("127.0.0.1:0.%s", port)) + return fakeMasterAddress(lis.Addr()) +} + +// fakeMasterAddress names a fake master by its listener port. The port doubles +// as the HTTP port so that ServerAddress.Equals, which compares HTTP +// addresses, tells two fake masters apart. +func fakeMasterAddress(addr net.Addr) pb.ServerAddress { + _, port, _ := net.SplitHostPort(addr.String()) + return pb.ServerAddress(fmt.Sprintf("127.0.0.1:%s.%s", port, port)) } // fakeHeartbeatMaster answers SendHeartbeat the way a master does: a follower @@ -311,8 +318,9 @@ func TestRefreshLeaderFromPeersPublishesReportedLeader(t *testing.T) { vs.store.MasterFn = vs.getCurrentMaster vs.setCurrentMaster("lost.example.com:9333.19333") - vs.refreshLeaderFromPeers(5 * time.Second) - + if got := vs.refreshLeaderFromPeers("lost.example.com:9333.19333", 5*time.Second); got != leader { + t.Fatalf("refresh returned %q, want reported leader %q", string(got), string(leader)) + } if got := vs.getCurrentMaster(); got != leader { t.Fatalf("currentMaster = %q, want reported leader %q", string(got), string(leader)) } @@ -333,8 +341,9 @@ func TestRefreshLeaderFromPeersClearsLeaderWithoutAnswer(t *testing.T) { vs.store.MasterFn = vs.getCurrentMaster vs.setCurrentMaster("lost.example.com:9333.19333") - vs.refreshLeaderFromPeers(5 * time.Second) - + if got := vs.refreshLeaderFromPeers("lost.example.com:9333.19333", 5*time.Second); got != "" { + t.Fatalf("refresh returned %q, want none", string(got)) + } if got := vs.getCurrentMaster(); got != "" { t.Fatalf("currentMaster = %q, want none", string(got)) } @@ -342,3 +351,183 @@ func TestRefreshLeaderFromPeersClearsLeaderWithoutAnswer(t *testing.T) { t.Fatalf("store master = %q, want none", string(got)) } } + +// startUnresponsiveMaster listens without ever answering, the way a master +// behind a partition or on a powered-off host looks to a dialer: no refusal, +// just a connection that never becomes usable. +func startUnresponsiveMaster(t *testing.T) pb.ServerAddress { + t.Helper() + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + t.Cleanup(func() { _ = lis.Close() }) + return fakeMasterAddress(lis.Addr()) +} + +// An unreachable lost master listed first must not use up the refresh before +// a live peer is asked who leads now. +func TestRefreshLeaderFromPeersAsksLostMasterLast(t *testing.T) { + const leader = pb.ServerAddress("leader.example.com:9333.19333") + lost := startUnresponsiveMaster(t) + peer := startFakeMasterServerForLeaderLookup(t, &fakeLeaderConfigServer{leader: string(leader)}) + vs := &VolumeServer{ + SeedMasterNodes: []pb.ServerAddress{lost, peer}, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + store: &storage.Store{Ip: "127.0.0.1", Port: 8080}, + } + vs.setCurrentMaster(lost) + + if got := vs.refreshLeaderFromPeers(lost, 2*time.Second); got != leader { + t.Fatalf("refresh returned %q, want %q from the live peer", string(got), string(leader)) + } + if got := vs.getCurrentMaster(); got != leader { + t.Fatalf("currentMaster = %q, want %q", string(got), string(leader)) + } +} + +func TestMastersAskingLast(t *testing.T) { + a, b, c := pb.ServerAddress("a:9333"), pb.ServerAddress("b:9333"), pb.ServerAddress("c:9333") + got := mastersAskingLast([]pb.ServerAddress{a, b, c}, "a:9333.19333") + want := []pb.ServerAddress{b, c, a} + if fmt.Sprint(got) != fmt.Sprint(want) { + t.Fatalf("got %v, want %v", got, want) + } + if got := mastersAskingLast([]pb.ServerAddress{a, b}, ""); fmt.Sprint(got) != fmt.Sprint([]pb.ServerAddress{a, b}) { + t.Fatalf("no lost master: got %v", got) + } +} + +// switchableFollower names whichever master it currently believes leads, on +// heartbeat streams and in its configuration alike. +type switchableFollower struct { + master_pb.UnimplementedSeaweedServer + mu sync.Mutex + leader pb.ServerAddress +} + +func (f *switchableFollower) setLeader(leader pb.ServerAddress) { + f.mu.Lock() + f.leader = leader + f.mu.Unlock() +} + +func (f *switchableFollower) currentLeader() string { + f.mu.Lock() + defer f.mu.Unlock() + return string(f.leader) +} + +func (f *switchableFollower) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServer) error { + for { + if _, err := stream.Recv(); err != nil { + return err + } + if err := stream.Send(&master_pb.HeartbeatResponse{Leader: f.currentLeader()}); err != nil { + return err + } + } +} + +func (f *switchableFollower) GetMasterConfiguration(_ context.Context, _ *master_pb.GetMasterConfigurationRequest) (*master_pb.GetMasterConfigurationResponse, error) { + return &master_pb.GetMasterConfigurationResponse{Leader: f.currentLeader()}, nil +} + +// When the leader's stream breaks and the leader stays unreachable, the next +// stream goes to the leader the other masters name, not to the next seed. The +// seeds are ordered so that the next seed is the lost leader itself, which is +// how a volume server that reached the old leader through a redirect finds +// its seed list after the failure. +func TestHeartbeatAfterLosingLeaderDialsTheNewLeader(t *testing.T) { + oldLeader := &fakeHeartbeatMaster{heartbeats: make(chan *master_pb.Heartbeat, 8)} + oldLis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + oldServer := grpc.NewServer() + master_pb.RegisterSeaweedServer(oldServer, oldLeader) + go func() { _ = oldServer.Serve(oldLis) }() + t.Cleanup(oldServer.Stop) + oldAddr := fakeMasterAddress(oldLis.Addr()) + + newLeader := &fakeHeartbeatMaster{heartbeats: make(chan *master_pb.Heartbeat, 8)} + newAddr := startFakeMasterServerForLeaderLookup(t, newLeader) + follower := &switchableFollower{leader: oldAddr} + followerAddr := startFakeMasterServerForLeaderLookup(t, follower) + + vs := newHeartbeatTestVolumeServer(t) + vs.SeedMasterNodes = []pb.ServerAddress{followerAddr, newAddr, oldAddr} + vs.pulsePeriod = time.Second + exited := make(chan struct{}) + go func() { + defer close(exited) + vs.heartbeat() + }() + t.Cleanup(func() { + vs.StopHeartbeat() + select { + case <-exited: + case <-time.After(10 * time.Second): + t.Error("heartbeat loop did not stop") + } + }) + + select { + case <-oldLeader.heartbeats: + case <-time.After(10 * time.Second): + t.Fatal("old leader got no heartbeat") + } + + // The old leader becomes unreachable: its streams break, and its port + // accepts connections that never answer. + follower.setLeader(newAddr) + oldServer.Stop() + blackHole, err := net.Listen("tcp", oldLis.Addr().String()) + if err != nil { + t.Fatalf("reopen %s: %v", oldLis.Addr(), err) + } + t.Cleanup(func() { _ = blackHole.Close() }) + + // A dial to the unreachable old leader alone would take the gRPC connect + // timeout (20s). + select { + case <-newLeader.heartbeats: + case <-time.After(10 * time.Second): + t.Fatalf("new leader got no heartbeat within 10s of losing the old one; currentMaster = %q", vs.getCurrentMaster()) + } +} + +// With a single seed the hint budget is one try per pass. A seed stuck on a +// stale name spends it every pass, so its answer must not drop the live name +// that answer-chain reached before it was tried. +func TestHeartbeatKeepsUntriedHintOverStaleSeedAnswer(t *testing.T) { + live := &fakeHeartbeatMaster{heartbeats: make(chan *master_pb.Heartbeat, 8)} + liveAddr := startFakeMasterServerForLeaderLookup(t, live) + redirects := &fakeHeartbeatMaster{leader: string(liveAddr)} + redirectAddr := startFakeMasterServerForLeaderLookup(t, redirects) + stale := &fakeHeartbeatMaster{leader: string(redirectAddr)} + staleAddr := startFakeMasterServerForLeaderLookup(t, stale) + + vs := newHeartbeatTestVolumeServer(t) + vs.SeedMasterNodes = []pb.ServerAddress{staleAddr} + vs.pulsePeriod = time.Second + exited := make(chan struct{}) + go func() { + defer close(exited) + vs.heartbeat() + }() + t.Cleanup(func() { + vs.StopHeartbeat() + select { + case <-exited: + case <-time.After(10 * time.Second): + t.Error("heartbeat loop did not stop") + } + }) + + select { + case <-live.heartbeats: + case <-time.After(15 * time.Second): + t.Fatalf("live leader got no heartbeat; currentMaster = %q", vs.getCurrentMaster()) + } +}