volume: redial a master that just dropped a registered stream after a second (#12420)

* volume: redial a master that just dropped a registered stream after a second

A registered volume server that loses its master with no peer naming a
leader waited a full pulse before redialing. A restarted master is
usually back within a second, so its volumes stayed unregistered, and
uncached lookups kept failing, for most of a pulse.

For one pulse after a stream a leader had answered ends, redial on the
one-second cap. A master still down after that is redialed at the pulse
pace, so redials never come later than before.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* seaweed-volume: redial a master that just dropped a registered stream after a second

Mirror of the Go heartbeat change. The loop keeps when the last stream a
leader had answered ended, which also stands in for the
confirmed-leader-once flag (None until first registration), and redials
on the one-second cap for a pulse after that.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* seaweed-volume: test that a dropped registered stream records the loss

A fake master confirms the stream and then ends it. The server starts as
registered long ago, so only a recorded loss makes the redial come after
a second instead of a pulse.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* volume: record a lost stream only after its receiver has exited

doHeartbeatWithRetry can return on a send error while the receiver is
still handling the leader's first reply, so the deferred recorder could
read the registered flag too early and keep an old loss time. Cancel the
stream and wait for the receiver before reading it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test: require the prompt heartbeat redial to wait out its second

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* seaweed-volume: fail the down master's dials at once in the redial test

Windows takes about two seconds to refuse a connection to a closed
loopback port, so the stream dial and the leader lookup each held the
prompt case past its two-second bound. A listener that closes every
connection fails both dials immediately on every platform.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

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<String> = 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<tokio::time::Instant> = None;
 
     loop {
         let mut tried_hints: HashSet<String> = 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<tokio::time::Instant>,
     duplicate_retry_count: &mut u32,
 ) -> Option<String> {
     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<Option<String>, Box<dyn std::error::Error>> {
     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<tonic::codegen::http::Request<tonic::body::Body>> for DroppingLeader {
+        type Response = tonic::codegen::http::Response<tonic::body::Body>;
+        type Error = std::convert::Infallible;
+        type Future = tonic::codegen::BoxFuture<Self::Response, Self::Error>;
+
+        fn poll_ready(
+            &mut self,
+            _cx: &mut std::task::Context<'_>,
+        ) -> std::task::Poll<Result<(), Self::Error>> {
+            std::task::Poll::Ready(Ok(()))
+        }
+
+        fn call(&mut self, req: tonic::codegen::http::Request<tonic::body::Body>) -> 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<master_pb::Heartbeat> for ConfirmThenDrop {
+        type Response = master_pb::HeartbeatResponse;
+        type ResponseStream = tokio_stream::Iter<
+            std::vec::IntoIter<Result<master_pb::HeartbeatResponse, tonic::Status>>,
+        >;
+        type Future =
+            tonic::codegen::BoxFuture<tonic::Response<Self::ResponseStream>, tonic::Status>;
+
+        fn call(
+            &mut self,
+            request: tonic::Request<tonic::Streaming<master_pb::Heartbeat>>,
+        ) -> 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]

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

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)
+	}
+}

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