diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index 785df66dd0..dde4d7106b 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/seaweed-volume/src/server/heartbeat.rs @@ -64,10 +64,11 @@ pub async fn run_heartbeat_with_state( let pulse = Duration::from_secs(config.pulse_seconds.max(1)); let mut new_leader: Option = None; let mut duplicate_retry_count: u32 = 0; - // Whether a heartbeat stream was ever answered by a leader; before first - // registration the retry loop reconnects promptly instead of pacing a - // full pulse. Mirrors Go's `confirmedLeaderOnce`. - let mut confirmed_leader_once = false; + // When a stream a leader had answered last ended, `None` before first + // registration. Until then, and for a pulse after each loss, the retry + // loop reconnects promptly instead of pacing a full pulse. Mirrors Go's + // `confirmedLeaderOnce` and `leaderLostAt`. + let mut leader_lost_at: Option = None; loop { let mut tried_hints: HashSet = HashSet::new(); @@ -110,7 +111,7 @@ pub async fn run_heartbeat_with_state( &leader, pulse, &mut shutdown_rx, - &mut confirmed_leader_once, + &mut leader_lost_at, &mut duplicate_retry_count, ) .await; @@ -134,7 +135,7 @@ pub async fn run_heartbeat_with_state( master_addr, pulse, &mut shutdown_rx, - &mut confirmed_leader_once, + &mut leader_lost_at, &mut duplicate_retry_count, ) .await @@ -168,7 +169,7 @@ async fn heartbeat_once( target_addr: &str, pulse: Duration, shutdown_rx: &mut broadcast::Receiver<()>, - confirmed_leader_once: &mut bool, + leader_lost_at: &mut Option, duplicate_retry_count: &mut u32, ) -> Option { let grpc_addr = to_grpc_address(target_addr); @@ -181,17 +182,22 @@ async fn heartbeat_once( SleepDuplicate(Duration), SleepPulse, } - let action = match do_heartbeat( + let mut confirmed_leader = false; + let result = do_heartbeat( config, state, &grpc_addr, target_addr, pulse, shutdown_rx, - confirmed_leader_once, + &mut confirmed_leader, ) .await - { + .map_err(|e| e.to_string()); + if confirmed_leader { + *leader_lost_at = Some(tokio::time::Instant::now()); + } + let action = match result { Ok(Some(leader)) => { info!("Master leader changed to {}", leader); return Some(leader); @@ -200,10 +206,7 @@ async fn heartbeat_once( *duplicate_retry_count = 0; return None; } - Err(e) => { - let err_msg = e.to_string(); - // Drop `e` (non-Send) before any .await - drop(e); + Err(err_msg) => { warn!("Heartbeat to {} error: {}", grpc_addr, err_msg); if err_msg.contains(DUPLICATE_UUID_RETRY_MESSAGE) { @@ -236,16 +239,17 @@ async fn heartbeat_once( // The lookup spends part of the retry delay rather than adding // to it. let retry_at = tokio::time::Instant::now() + pulse; + let unregistered_or_just_lost = leader_lost_at.is_none_or(|at| at.elapsed() < 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) + && (!state.current_master().await.is_empty() || unregistered_or_just_lost) { // 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. + // next stream right away, an unregistered server has + // nothing else to do, and a master that just dropped the + // stream is usually back within a second, so reconnect + // promptly. Otherwise keep the pulse backoff. wait = Duration::from_secs(1); } tokio::time::sleep(wait).await; @@ -564,7 +568,7 @@ async fn do_heartbeat( current_master: &str, pulse: Duration, shutdown_rx: &mut broadcast::Receiver<()>, - confirmed_leader_once: &mut bool, + confirmed_leader: &mut bool, ) -> Result, Box> { let channel = connect_channel( grpc_addr, @@ -611,7 +615,6 @@ async fn do_heartbeat( volume_tick.tick().await; ec_tick.tick().await; - let mut confirmed_leader = false; loop { tokio::select! { resp = response_stream.message() => { @@ -629,12 +632,11 @@ async fn do_heartbeat( publish_current_master(state, &hb_resp.leader).await; return Ok(Some(hb_resp.leader)); } - } else if !confirmed_leader { + } else if !*confirmed_leader { // The leader answers every new stream without // naming a leader, so this peer is the one lookups // go to. - confirmed_leader = true; - *confirmed_leader_once = true; + *confirmed_leader = true; publish_current_master(state, current_master).await; } if !hb_resp.duplicated_uuids.is_empty() { @@ -2428,6 +2430,169 @@ mod tests { assert!(volume.is_no_write_or_delete()); } + // A master that is down is redialed after about a second for a pulse after + // the stream to it was lost, since a restarting master is usually back by + // then, and at the pulse pace once the loss is older. Mirrors Go's + // TestHeartbeatOnceKeepsPulseBackoffOnceLossIsOld. + #[tokio::test] + async fn test_heartbeat_once_redials_promptly_only_for_a_pulse_after_a_loss() { + // Closing each connection fails every dial at once, where Windows takes + // about two seconds to refuse a closed port. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + tokio::spawn(async move { + while let Ok((conn, _)) = listener.accept().await { + drop(conn); + } + }); + let down = format!("127.0.0.1:{}.{}", port, port); + let pulse = Duration::from_secs(3); + let mut config = test_config(); + config.master_addresses = vec![down.clone()]; + config.pulse_seconds = pulse.as_secs(); + let temp_dir = tempfile::tempdir().unwrap(); + let state = test_state_with_store(reporting_store(temp_dir.path().to_str().unwrap(), 0)); + let (_shutdown_tx, mut shutdown_rx) = broadcast::channel(1); + + for (lost_ago, prompt) in [(Duration::ZERO, true), (pulse, false)] { + let mut leader_lost_at = Some(tokio::time::Instant::now() - lost_ago); + let mut duplicate_retry_count = 0; + let started = tokio::time::Instant::now(); + heartbeat_once( + &config, + &state, + &down, + pulse, + &mut shutdown_rx, + &mut leader_lost_at, + &mut duplicate_retry_count, + ) + .await; + let elapsed = started.elapsed(); + if prompt { + assert!( + elapsed >= Duration::from_millis(900) && elapsed < Duration::from_secs(2), + "redial right after a loss took {:?}", + elapsed + ); + } else { + assert!( + elapsed >= Duration::from_millis(2500), + "redial a pulse after a loss took {:?}", + elapsed + ); + } + } + } + + /// A master that confirms each heartbeat stream and then ends it, the way + /// one killed right after registering this server looks. It answers + /// nothing else, so no peer names a leader. + #[derive(Clone)] + struct DroppingLeader; + + impl tonic::server::NamedService for DroppingLeader { + const NAME: &'static str = "master_pb.Seaweed"; + } + + impl tonic::codegen::Service> for DroppingLeader { + type Response = tonic::codegen::http::Response; + type Error = std::convert::Infallible; + type Future = tonic::codegen::BoxFuture; + + fn poll_ready( + &mut self, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Ok(())) + } + + fn call(&mut self, req: tonic::codegen::http::Request) -> Self::Future { + Box::pin(async move { + if req.uri().path() != "/master_pb.Seaweed/SendHeartbeat" { + return Ok(tonic::Status::unimplemented("").into_http()); + } + let mut grpc = tonic::server::Grpc::new(tonic_prost::ProstCodec::default()); + Ok(grpc.streaming(ConfirmThenDrop, req).await) + }) + } + } + + struct ConfirmThenDrop; + + impl tonic::server::StreamingService for ConfirmThenDrop { + type Response = master_pb::HeartbeatResponse; + type ResponseStream = tokio_stream::Iter< + std::vec::IntoIter>, + >; + type Future = + tonic::codegen::BoxFuture, tonic::Status>; + + fn call( + &mut self, + request: tonic::Request>, + ) -> Self::Future { + Box::pin(async move { + let _ = request.into_inner().message().await; + Ok(tonic::Response::new(tokio_stream::iter(vec![ + Ok(master_pb::HeartbeatResponse { + volume_size_limit: 1 << 30, + ..Default::default() + }), + Err(tonic::Status::unavailable("master stopped")), + ]))) + }) + } + } + + // A server registered long ago whose master just dropped the stream + // redials after about a second, not a pulse. Mirrors Go's + // TestHeartbeatOnceRedialsPromptlyAfterLosingRegisteredStream. + #[tokio::test] + async fn test_heartbeat_once_redials_promptly_after_losing_a_registered_stream() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let master = format!("127.0.0.1:{}.{}", port, port); + tokio::spawn( + tonic::transport::Server::builder() + .add_service(DroppingLeader) + .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener)), + ); + let pulse = Duration::from_secs(3); + let mut config = test_config(); + config.master_addresses = vec![master.clone()]; + config.pulse_seconds = pulse.as_secs(); + let temp_dir = tempfile::tempdir().unwrap(); + let state = test_state_with_store(reporting_store(temp_dir.path().to_str().unwrap(), 0)); + let (_shutdown_tx, mut shutdown_rx) = broadcast::channel(1); + let registered_long_ago = tokio::time::Instant::now() - 10 * pulse; + let mut leader_lost_at = Some(registered_long_ago); + let mut duplicate_retry_count = 0; + + let started = tokio::time::Instant::now(); + heartbeat_once( + &config, + &state, + &master, + pulse, + &mut shutdown_rx, + &mut leader_lost_at, + &mut duplicate_retry_count, + ) + .await; + let elapsed = started.elapsed(); + + assert!( + leader_lost_at.is_some_and(|at| at >= started), + "the end of the confirmed stream was not recorded" + ); + assert!( + elapsed >= Duration::from_millis(900) && elapsed < Duration::from_secs(2), + "redial after losing a registered stream took {:?}", + elapsed + ); + } + // The pass only reads the store, so it must not shut out the readers that // serve traffic while it stats every volume. #[tokio::test] diff --git a/weed/server/volume_grpc_client_to_master.go b/weed/server/volume_grpc_client_to_master.go index bb48e6f14c..4e5c45f942 100644 --- a/weed/server/volume_grpc_client_to_master.go +++ b/weed/server/volume_grpc_client_to_master.go @@ -6,6 +6,7 @@ import ( "fmt" "os" "strings" + "sync/atomic" "time" "github.com/seaweedfs/seaweedfs/weed/operation" @@ -231,14 +232,15 @@ func (vs *VolumeServer) heartbeatOnce(target pb.ServerAddress, grpcDialOption gr // The lookup spends part of the retry delay rather than // adding to it. retryAt := time.Now().Add(vs.pulsePeriod) + lostRecently := time.Since(time.Unix(0, vs.leaderLostAt.Load())) < vs.pulsePeriod leader = vs.refreshLeaderFromPeers(target, vs.pulsePeriod) wait := time.Until(retryAt) - if wait > time.Second && (vs.getCurrentMaster() != "" || !vs.confirmedLeaderOnce.Load()) { + if wait > time.Second && (vs.getCurrentMaster() != "" || !vs.confirmedLeaderOnce.Load() || lostRecently) { // 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. + // next stream right away, an unregistered server has + // nothing else to do, and a master that just dropped the + // stream is usually back within a second, so reconnect + // promptly. Otherwise keep the pulse backoff. wait = time.Second } time.Sleep(wait) @@ -288,8 +290,18 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp // since a gRPC stream allows only one sender at a time. resendChan := make(chan struct{}, 1) + var registered atomic.Bool + recvDone := make(chan struct{}) + defer func() { + cancel() + <-recvDone + if registered.Load() { + vs.leaderLostAt.Store(time.Now().UnixNano()) + } + }() + go func() { - confirmedLeader := false + defer close(recvDone) for { in, err := stream.Recv() if err != nil { @@ -312,10 +324,10 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp doneChan <- nil return } - } else if !confirmedLeader { + } else if !registered.Load() { // The leader answers every new heartbeat stream without naming a // leader, so this peer is the one lookups should go to. - confirmedLeader = true + registered.Store(true) vs.confirmedLeaderOnce.Store(true) vs.setCurrentMaster(masterAddress) } diff --git a/weed/server/volume_grpc_client_to_master_test.go b/weed/server/volume_grpc_client_to_master_test.go index e1c55e6d68..1c17348bfe 100644 --- a/weed/server/volume_grpc_client_to_master_test.go +++ b/weed/server/volume_grpc_client_to_master_test.go @@ -531,3 +531,62 @@ func TestHeartbeatKeepsUntriedHintOverStaleSeedAnswer(t *testing.T) { t.Fatalf("live leader got no heartbeat; currentMaster = %q", vs.getCurrentMaster()) } } + +// droppingLeader confirms each heartbeat stream and then ends it, the way a +// master that is killed right after registering a volume server looks. +type droppingLeader struct { + master_pb.UnimplementedSeaweedServer +} + +func (m *droppingLeader) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServer) error { + if _, err := stream.Recv(); err != nil { + return err + } + if err := stream.Send(&master_pb.HeartbeatResponse{VolumeSizeLimit: 1 << 30}); err != nil { + return err + } + return fmt.Errorf("master stopped") +} + +func timeHeartbeatOnce(vs *VolumeServer, target pb.ServerAddress) time.Duration { + start := time.Now() + vs.heartbeatOnce(target, vs.grpcDialOption, 0) + return time.Since(start) +} + +// A registered server whose master just dropped the stream redials after about +// a second, since a restarting master is usually back by then, even though no +// peer names a leader. +func TestHeartbeatOnceRedialsPromptlyAfterLosingRegisteredStream(t *testing.T) { + addr := startFakeMasterServerForLeaderLookup(t, &droppingLeader{}) + vs := newHeartbeatTestVolumeServer(t) + vs.SeedMasterNodes = []pb.ServerAddress{addr} + vs.pulsePeriod = 3 * time.Second + + if elapsed := timeHeartbeatOnce(vs, addr); elapsed < 900*time.Millisecond || elapsed >= 2*time.Second { + t.Fatalf("redial after losing a registered stream took %v, want about 1s", elapsed) + } + if !vs.confirmedLeaderOnce.Load() { + t.Fatal("the dropping leader never confirmed the stream") + } +} + +// Once the loss is a pulse old, a master that is still down is redialed at the +// pulse pace again. +func TestHeartbeatOnceKeepsPulseBackoffOnceLossIsOld(t *testing.T) { + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + down := fakeMasterAddress(lis.Addr()) + lis.Close() + vs := newHeartbeatTestVolumeServer(t) + vs.SeedMasterNodes = []pb.ServerAddress{down} + vs.pulsePeriod = 3 * time.Second + vs.confirmedLeaderOnce.Store(true) + vs.leaderLostAt.Store(time.Now().Add(-vs.pulsePeriod).UnixNano()) + + if elapsed := timeHeartbeatOnce(vs, down); elapsed < 2500*time.Millisecond { + t.Fatalf("redial of a master lost a pulse ago took %v, want the %v pulse", elapsed, vs.pulsePeriod) + } +} diff --git a/weed/server/volume_server.go b/weed/server/volume_server.go index bb5ab57016..7d1ac9a3e5 100644 --- a/weed/server/volume_server.go +++ b/weed/server/volume_server.go @@ -53,6 +53,11 @@ type VolumeServer struct { guard *security.Guard grpcDialOption grpc.DialOption + // leaderLostAt is when a stream a leader had answered last ended, in Unix + // nanoseconds. For a pulse after that the loop also reconnects promptly: + // a master that just dropped the stream is usually restarting. + leaderLostAt atomic.Int64 + needleMapKind storage.NeedleMapKind ldbTimout int64 FixJpgOrientation bool