volume: route lookups to the leader master only (#11888)

* volume: route lookups to the leader master only (#11887) After a leader change, the volume server kept sending volume and EC shard lookups to masters that were not the leader: * currentMaster was set as soon as a heartbeat stream opened, before the peer said whether it was leader, and kept a follower through the redirect and its 3s sleep. A follower answers LookupVolume from its own vid cache, which right after an election can be empty, so reads went 404 rather than failing over. * EC lookups used Store.MasterAddress, a second, unlocked copy that the heartbeat loop set before each dial and cleared on errors, so EC and plain reads could disagree on the master or have none. * the receive goroutine called stream.Send itself when the volume options changed, racing the send loop, and exited silently on failure, after which leader redirects were no longer read. Publish currentMaster only once a master is known to be leader: the peer answered a new stream without naming a leader, which the leader always does, or a follower named the leader. Compare redirects against the connected peer. Replace Store.MasterAddress with MasterFn, backed by getCurrentMaster. Have the receiver ask the send loop for the extra heartbeat instead of sending it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume: ask the masters for the leader after losing the heartbeat When the heartbeat stream to the leader failed, currentMaster kept the lost master until a new stream answered, so lookups on a cache miss kept dialing it through the retry delay even with a new leader already elected. On a regular heartbeat error, ask the configured masters for the raft leader and publish it, within the retry delay rather than on top of it. Without an answer the last known leader stays: clearing it would only trade a dead address for an empty one. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test: stop the heartbeat on a failed assertion The leader test closed stopChan only once every assertion had passed, so a failure left the stream open and the fake master's GracefulStop waited for the whole test run's timeout. Start the heartbeat through a helper whose cleanup stops it, safe to call twice, and waits for it to exit. It runs before the cleanups registered earlier, so the stream is closed before the store and the fake master go away. Both heartbeat tests use it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume: clear the leader when no master names one When the heartbeat lost its master and no configured master named a leader, refreshLeaderFromPeers kept the lost one. That master is dead or no longer leader, and a follower answers LookupVolume from its own cache instead of refusing, so lookups could return locations no leader vouched for. Clear currentMaster instead. Lookups then fail fast ("missing address") until a leader is known again, rather than reading a follower's cache or waiting on a dial to a dead master. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume: keep volumeSizeLimit 64-bit aligned on 32-bit builds MasterFn (a func, 4 bytes on 386) replaced MasterAddress (a string, 8 bytes) ahead of volumeSizeLimit and moved it to a 4-byte-aligned offset, where atomic.LoadUint64 panics with an unaligned 64-bit operation. The atomic.Uint64 type carries the alignment itself. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume: publish the answering leader before a duplicate-uuid retry Read the leader field first on each heartbeat reply. A follower reply carries nothing else, so it no longer runs through the option updates; and a leader that rejects the registration over duplicate uuids still answers lookups, so it stays the lookup target through the retry wait instead of leaving the previous master in place. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume: publish the master only once it is known leader (rust) * volume: ask the masters for the leader after losing the heartbeat (rust) * volume: read the leader field first on each heartbeat reply (rust) * volume: drop the seed fallback from ec shard lookups (rust) cached_lookup_ec_shard_locations fell back to the static first seed when no leader was published, so a mid-election EC lookup could land on a follower and read its stale shard map. Error out like Go's masterAddress() returning "" instead, via a current_master() getter matching getCurrentMaster. * volume: send volume lookups to the confirmed leader, not the seed (rust) Every HTTP lookup path — proxy/redirect targets, replicated writes and deletes, batch chunk deletes, remote chunk reads — dialed the static first seed, so after a failover they kept asking a follower whose vid map can be stale. Go routes all of these through GetMaster; switch the lookup calls and their leader-known gates to current_master(). * volume: mark volumes readonly on the leader, retrying the named one (rust) notify_master_volume_readonly fell back to the seed when no leader was published, which the peer answers with "Not current leader". It now uses only the confirmed leader, skips the call in masterless mode, and on NotLeader asks the seed masters who leads and retries there — the same path Go's notifyMasterVolumeReadonly takes. The peer leader query is shared with refresh_leader_from_peers as lookup_leader_from_peers. * volume: ask the confirmed leader for copy prealloc options (rust) VolumeCopy's preallocate query went to the static seed; Go issues it through GetMaster. Route it through current_master() too — when no leader is known the lookup fails fast and the copy proceeds without preallocating, as before. * volume: keep the announced master port forms, not just http (rust) publish_current_master stripped to http form, so a leader configured with an explicit gRPC port (host:9333.19999) was redialed on the default offset. Store the announced host:port[.grpcPort] form like Go's currentMaster; callers pick to_http_address / to_grpc_address / parse_grpc_address per use, and Ping admission compares on the http form. * volume: fail replicated writes and deletes when no leader is known (rust) Gating replication on a known leader meant a write could be acknowledged on fewer copies than required, and a delete could leave the data readable on the other replicas. Go fails the request when the replica-location lookup cannot run; drop the gate so the empty-leader lookup errors the same way. --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
2026-10-22 22:11:39 +02:00 · 2026-10-21 13:36:05 -07:00
1 parent 4bdbfbaaa8
commit 23d8c62844
@@ -53,6 +53,12 @@ fn needle_not_found(needle_id: NeedleId) -> Status {
    Status::not_found(format!("needle not found {}", needle_id.0))
 }

+/// Mirrors Go's `isNotLeaderErr`: the master returns raft's NotLeaderError,
+/// which arrives as an Unknown-status message.
+fn is_not_leader_status(e: &Status) -> bool {
+    e.message().contains("Not current leader")
+}
+
 /// Deadline from the caller's `grpc-timeout` header (TimeoutValue +
 /// H|M|S|m|u|n unit, per the gRPC-over-HTTP2 spec). Tonic's GrpcTimeout layer
 /// drops an over-deadline handler future anyway; parsing it here lets a queued
@@ -344,22 +350,46 @@ impl VolumeGrpcService {
        is_readonly: bool,
    ) -> Result<(), Status> {
        // VolumeMarkReadonly mutates raft-replicated master topology, so it must
-        // reach the leader. Prefer the live leader the heartbeat is talking to
-        // (current_master_url), falling back to the static seed before the first
-        // heartbeat — mirrors store_ec.rs and Go's vs.GetMaster(). Sending it to
-        // the static seed fails "not current leader" after any master failover.
-        let master_url = {
-            let live = self.state.current_master_url.read().await.clone();
-            if !live.is_empty() {
-                live
-            } else {
-                self.state.master_url.clone()
-            }
-        };
+        // reach the leader the heartbeat confirmed. On "Not current leader",
+        // ask the seed masters who leads and retry there — mirrors Go's
+        // notifyMasterVolumeReadonly.
+        let master_url = self.state.current_master().await;
        if master_url.is_empty() {
-            return Ok(());
+            if !self.state.has_master {
+                return Ok(());
+            }
+            return Err(Status::unavailable("no master leader known"));
        }
-        let grpc_addr = parse_grpc_address(&master_url).map_err(|e| {
+        match self
+            .volume_mark_readonly_on_master(&master_url, info, is_readonly)
+            .await
+        {
+            Err(e) if is_not_leader_status(&e) => {
+                let leader = super::heartbeat::lookup_leader_from_peers(
+                    &self.state.master_urls,
+                    self.state.outgoing_grpc_tls.as_ref(),
+                )
+                .await
+                .ok_or_else(|| {
+                    Status::unavailable(format!(
+                        "master {} rejected mark-readonly and no peer named a leader",
+                        master_url
+                    ))
+                })?;
+                self.volume_mark_readonly_on_master(&leader, info, is_readonly)
+                    .await
+            }
+            result => result,
+        }
+    }
+
+    async fn volume_mark_readonly_on_master(
+        &self,
+        master_url: &str,
+        info: &MasterVolumeInfo,
+        is_readonly: bool,
+    ) -> Result<(), Status> {
+        let grpc_addr = parse_grpc_address(master_url).map_err(|e| {
            Status::internal(format!("invalid master address {}: {}", master_url, e))
        })?;
        let channel = connect_channel(
@@ -5776,7 +5806,8 @@ impl VolumeCopyJob {
            (&self.dest.data_base_name, self.dest.has_remote_dat);
        let mut preallocate_size: i64 = 0;
        if !has_remote_dat {
-            let grpc_addr = super::heartbeat::to_grpc_address(&state.master_url);
+            let master = state.current_master().await;
+            let grpc_addr = super::heartbeat::to_grpc_address(&master);
            // Race the master-configuration RPC against the caller's
            // response channel: a stalled master (or a slow leader
            // election) would otherwise hold the task and its .note
@@ -5801,7 +5832,7 @@ impl VolumeCopyJob {
                    }
                }
                Err(e) => {
-                    tracing::warn!("get master {} configuration: {}", state.master_url, e);
+                    tracing::warn!("get master {} configuration: {}", master, e);
                }
            }

@@ -603,7 +603,7 @@ async fn batch_delete_file_ids(
            let locations = lookup_volume(
                &state.http_client,
                &state.outgoing_http_scheme,
-                &state.master_url,
+                &state.current_master().await,
                volume_id,
            )
            .await
@@ -675,7 +675,7 @@ async fn do_replicated_request(
    let locations = lookup_volume(
        &state.http_client,
        &state.outgoing_http_scheme,
-        &state.master_url,
+        &state.current_master().await,
        vid,
    )
    .await
@@ -832,7 +832,7 @@ async fn proxy_or_redirect_to_target(
    let locations = match lookup_volume(
        &state.http_client,
        &state.outgoing_http_scheme,
-        &state.master_url,
+        &state.current_master().await,
        vid.0,
    )
    .await
@@ -1238,7 +1238,7 @@ async fn proxy_missing_volume(
    let query_string = uri.query().unwrap_or("").to_string();
    if is_proxied_query(&query_string)
        || state.read_mode == ReadMode::Local
-        || state.master_url.is_empty()
+        || state.current_master().await.is_empty()
    {
        return StatusCode::NOT_FOUND.into_response();
    }
@@ -1282,7 +1282,7 @@ async fn check_download_limit(
        // Go tries proxy to replica once before the wait loop
        // (checkDownloadLimit); it does not retry on each wakeup.
        let should_try_replica =
-            !is_proxied_query(&query_string) && !state.master_url.is_empty() && {
+            !is_proxied_query(&query_string) && !state.current_master().await.is_empty() && {
                let store = state.store.read().unwrap();
                store
                    .find_volume(vid)
@@ -2960,7 +2960,9 @@ pub async fn post_handler(

    // Replicate to remote volume servers if this volume has replicas.
    // Matches Go's GetWritableRemoteReplications: skip if copy_count == 1.
-    if !is_replicate && write_result.is_ok() && !state.master_url.is_empty() {
+    // No leader-known gate: when the lookup cannot run, the write must fail
+    // rather than be acknowledged on fewer copies than required.
+    if !is_replicate && write_result.is_ok() {
        let needs_replication = {
            let store = state.store.read().unwrap();
            store
@@ -3358,7 +3360,7 @@ pub async fn delete_handler(
    };

    let is_replicate = del_query.split('&').any(|p| p == "type=replicate");
-    if !is_replicate && delete_result.is_ok() && !state.master_url.is_empty() {
+    if !is_replicate && delete_result.is_ok() {
        let needs_replication = {
            let store = state.store.read().unwrap();
            store
@@ -4232,7 +4234,7 @@ async fn read_remote_chunk_needle(
    let locations = lookup_volume(
        &state.http_client,
        &state.outgoing_http_scheme,
-        &state.master_url,
+        &state.current_master().await,
        vid.0,
    )
    .await?;
@@ -6671,7 +6673,8 @@ mod tests {
        let s = Arc::get_mut(&mut state).unwrap();
        // Empty self_url makes `url.contains(self_url)` true for every peer.
        s.self_url = "198.51.100.8:8080".to_string();
-        s.master_url = master_url;
+        s.master_url = master_url.clone();
+        s.current_master_url = tokio::sync::RwLock::new(master_url);
        state
    }

@@ -155,7 +155,11 @@ pub async fn run_heartbeat_with_state(
                }
                PostAction::SleepPulse => {
                    new_leader = None;
-                    tokio::time::sleep(pulse).await;
+                    // 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;
+                    tokio::time::sleep_until(retry_at).await;
                }
            }

@@ -254,10 +258,56 @@ pub async fn try_get_master_configuration(
    Ok(resp.into_inner())
 }

+/// Publish the leader the configured masters report, so lookups stop going
+/// 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.
+async fn refresh_leader_from_peers(
+    config: &HeartbeatConfig,
+    state: &Arc<VolumeServerState>,
+    timeout: Duration,
+) {
+    let lookup = lookup_leader_from_peers(
+        &config.master_addresses,
+        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(),
+    }
+}
+
+/// 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`.
+pub async fn lookup_leader_from_peers(
+    master_addrs: &[String],
+    tls: Option<&super::grpc_client::OutgoingGrpcTlsConfig>,
+) -> Option<String> {
+    for master_addr in master_addrs {
+        let grpc_addr = to_grpc_address(master_addr);
+        if let Ok(resp) = try_get_master_configuration(&grpc_addr, tls).await
+            && !resp.leader.is_empty()
+        {
+            return Some(resp.leader);
+        }
+    }
+    None
+}
+
 fn is_stopping(state: &VolumeServerState) -> bool {
    *state.is_stopping.read().unwrap()
 }

+/// Store the master lookups should go to, in the address form it was
+/// announced in — `host:port` or `host:port.grpcPort` — so an explicit gRPC
+/// port survives to the gRPC callers, mirroring Go's setCurrentMaster.
+/// The heartbeat calls this only for a master known to lead.
+async fn publish_current_master(state: &Arc<VolumeServerState>, master: &str) {
+    *state.current_master_url.write().await = master.to_string();
+}
+
 fn duplicate_uuid_retry_delay(retry_count: u32) -> Duration {
    Duration::from_secs((1u64 << retry_count) * 2)
 }
@@ -430,14 +480,6 @@ async fn do_heartbeat(
    let mut response_stream = client.send_heartbeat(stream).await?.into_inner();

    info!("Heartbeat stream established with {}", grpc_addr);
-    // Publish the master we're now talking to in canonical http host:port
-    // form so Ping admission can recognise it once a leader change moves us
-    // off the seed list. Mirrors Go's vs.setCurrentMaster(masterAddress).
-    {
-        let normalised = super::volume_server::to_http_address(current_master).into_owned();
-        let mut guard = state.current_master_url.write().await;
-        *guard = normalised;
-    }
    if is_stopping(state) {
        state.is_heartbeating.store(false, Ordering::Relaxed);
        send_deregister_heartbeat(config, state, &tx).await;
@@ -451,13 +493,31 @@ async fn do_heartbeat(
    volume_tick.tick().await;
    ec_tick.tick().await;

+    let mut confirmed_leader = false;
    loop {
        tokio::select! {
            resp = response_stream.message() => {
                match resp {
                    Ok(Some(hb_resp)) => {
-                        // Match Go ordering: DuplicatedUuids first, then volume
-                        // options, then leader redirect.
+                        // A reply names the leader or comes from it, so it
+                        // settles where lookups go before anything else on the
+                        // reply is handled: a follower reply carries no other
+                        // fields, and a leader stays the lookup target even
+                        // when it rejects this registration. Mirrors Go.
+                        if !hb_resp.leader.is_empty() {
+                            if super::volume_server::to_http_address(current_master)
+                                != super::volume_server::to_http_address(&hb_resp.leader)
+                            {
+                                publish_current_master(state, &hb_resp.leader).await;
+                                return Ok(Some(hb_resp.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;
+                            publish_current_master(state, current_master).await;
+                        }
                        if !hb_resp.duplicated_uuids.is_empty() {
                            let uuids = hb_resp.duplicated_uuids.clone();
                            let duplicate_dirs = off_runtime(config, state, move |_, state| {
@@ -502,11 +562,6 @@ async fn do_heartbeat(
                        if metrics_changed {
                            state.metrics_notify.notify_waiters();
                        }
-                        // Match Go: only redirect if leader is non-empty AND
-                        // different from the current master we're connected to.
-                        if !hb_resp.leader.is_empty() && current_master != hb_resp.leader {
-                            return Ok(Some(hb_resp.leader));
-                        }
                    }
                    Ok(None) => return Ok(None),
                    Err(e) => return Err(Box::new(e)),
@@ -1139,16 +1139,9 @@ async fn cached_lookup_ec_shard_locations(
    state: &Arc<VolumeServerState>,
    vid: VolumeId,
 ) -> io::Result<HashMap<ShardId, Vec<String>>> {
-    let master = {
-        let live = state.current_master_url.read().await.clone();
-        if !live.is_empty() {
-            live
-        } else {
-            state.master_url.clone()
-        }
-    };
+    let master = state.current_master().await;
    if master.is_empty() {
-        return Err(io::Error::other("no master configured for ec shard lookup"));
+        return Err(io::Error::other("no master leader known for ec shard lookup"));
    }

    let grpc_addr =
@@ -86,7 +86,8 @@ pub struct VolumeServerState {
    /// If true, FetchAndWriteNeedle skips remote S3 endpoint validation,
    /// allowing arbitrary (incl. loopback / link-local / metadata) hosts.
    pub allow_untrusted_remote_endpoints: bool,
-    /// First master address for volume lookups (e.g., "localhost:9333").
+    /// First configured master address (e.g., "localhost:9333"). A seed, not a
+    /// lookup target: lookups go to `current_master_url`.
    pub master_url: String,
    /// Seed master addresses for UI rendering.
    pub master_urls: Vec<String>,
@@ -94,10 +95,12 @@ pub struct VolumeServerState {
    /// Built once at construction so Ping admission stays O(1). Mirrors
    /// Go's `seedMasterSet` on `VolumeServer`.
    pub seed_master_set: std::collections::HashSet<String>,
-    /// Current master this server is heartbeating with, in canonical http
-    /// `host:port` form. Empty when no heartbeat connection is active. The
-    /// heartbeat goroutine writes; admission reads — the lock keeps them
-    /// from racing on a leader change. Mirrors Go's `currentMaster` plus
+    /// Master this server takes as leader, in its announced
+    /// `host:port[.grpcPort]` form — callers pick http or gRPC form per use.
+    /// Empty until the heartbeat confirms one — a peer that answered a
+    /// stream without naming a leader, or the leader a follower named — and
+    /// cleared again when no master reports one. The heartbeat writes;
+    /// admission and lookups read. Mirrors Go's `currentMaster` plus
    /// `currentMasterLock`.
    pub current_master_url: tokio::sync::RwLock<String>,
    /// This server's own address (ip:port) for filtering self from lookup results.
@@ -140,6 +143,13 @@ pub struct VolumeServerState {
 }

 impl VolumeServerState {
+    /// The master lookups should go to: the leader the heartbeat confirmed,
+    /// or empty while none is known, in its announced `host:port[.grpcPort]`
+    /// form. Mirrors Go's `vs.getCurrentMaster()`.
+    pub async fn current_master(&self) -> String {
+        self.current_master_url.read().await.clone()
+    }
+
    /// Check if the server is in maintenance mode; return gRPC error if so.
    pub fn check_maintenance(&self) -> Result<(), tonic::Status> {
        if self.maintenance.load(Ordering::Relaxed) {
@@ -209,7 +219,7 @@ impl VolumeServerState {
            return false;
        }
        let current = self.current_master_url.read().await.clone();
-        if !current.is_empty() && current == key {
+        if !current.is_empty() && to_http_address(&current) == key {
            return true;
        }
        self.seed_master_set.contains(&key)
@@ -54,6 +54,22 @@ func (vs *VolumeServer) lookupRaftLeaderMaster(ctx context.Context) (pb.ServerAd
 	return leader, nil
 }

+// 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) {
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+	leader, err := vs.lookupRaftLeaderMaster(ctx)
+	if err != nil {
+		vs.setCurrentMaster("")
+		return
+	}
+	vs.setCurrentMaster(leader)
+}
+
 func isContextDoneErr(err error) bool {
 	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
 		return true
@@ -74,7 +90,10 @@ func (vs *VolumeServer) getCurrentMaster() pb.ServerAddress {
 }

 // setCurrentMaster updates vs.currentMaster under a write lock. The
-// heartbeat goroutine calls this whenever it (re)connects to a master.
+// heartbeat goroutine calls this only with a master known to be the leader:
+// one that answered a heartbeat as leader, or the one a follower named, on
+// the heartbeat stream or, after the stream failed, in its configuration.
+// After a failed stream with no leader named, it is cleared to "".
 func (vs *VolumeServer) setCurrentMaster(master pb.ServerAddress) {
 	vs.currentMasterLock.Lock()
 	vs.currentMaster = master
@@ -122,7 +141,6 @@ func (vs *VolumeServer) heartbeat() {
 				time.Sleep(3 * time.Second)
 				master = newLeader
 			}
-			vs.store.MasterAddress = master
 			newLeader, err = vs.doHeartbeatWithRetry(master, grpcDialOption, vs.pulsePeriod, duplicateRetryCount)
 			if err != nil {
 				glog.V(0).Infof("heartbeat to %s error: %v", master, err)
@@ -136,12 +154,15 @@ func (vs *VolumeServer) heartbeat() {
 				} else {
 					// Regular error, reset duplicate retry count
 					duplicateRetryCount = 0
-					time.Sleep(vs.pulsePeriod)
+					// The lookup spends part of the retry delay rather than
+					// adding to it.
+					retryAt := time.Now().Add(vs.pulsePeriod)
+					vs.refreshLeaderFromPeers(vs.pulsePeriod)
+					time.Sleep(time.Until(retryAt))
 				}

 				stats.VolumeServerMasterDisconnections.WithLabelValues(master.String()).Inc()
 				newLeader = ""
-				vs.store.MasterAddress = ""
 			} else {
 				// Successful connection, reset retry count
 				duplicateRetryCount = 0
@@ -180,17 +201,42 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp
 		return "", err
 	}
 	glog.V(0).Infof("Heartbeat to: %v", masterAddress)
-	vs.setCurrentMaster(masterAddress)

 	doneChan := make(chan error, 1)
+	// The receiver asks for a fresh heartbeat here rather than sending it,
+	// since a gRPC stream allows only one sender at a time.
+	resendChan := make(chan struct{}, 1)

 	go func() {
+		confirmedLeader := false
 		for {
 			in, err := stream.Recv()
 			if err != nil {
 				doneChan <- err
 				return
 			}
+			// A reply names the leader or comes from it, so it settles where
+			// lookups go before anything else on the reply is handled: a
+			// follower reply carries no other fields, and a leader stays the
+			// lookup target even when it rejects this registration.
+			if in.GetLeader() != "" {
+				// Only a follower names the leader. Publish the named leader right
+				// away, so lookups stop landing on this follower while the loop
+				// reconnects.
+				leader := pb.ServerAddress(in.GetLeader())
+				if !masterAddress.Equals(leader) {
+					glog.V(0).Infof("Volume Server found a new master newLeader: %v instead of %v", leader, masterAddress)
+					vs.setCurrentMaster(leader)
+					newLeader = leader
+					doneChan <- nil
+					return
+				}
+			} else if !confirmedLeader {
+				// The leader answers every new heartbeat stream without naming a
+				// leader, so this peer is the one lookups should go to.
+				confirmedLeader = true
+				vs.setCurrentMaster(masterAddress)
+			}
 			if len(in.DuplicatedUuids) > 0 {
 				var duplicateDir []string
 				for _, loc := range vs.store.Locations {
@@ -231,9 +277,9 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp
 			}
 			if volumeOptsChanged {
 				if vs.store.MaybeAdjustVolumeMax() {
-					if err = stream.Send(vs.store.CollectHeartbeat()); err != nil {
-						glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", vs.getCurrentMaster(), err)
-						return
+					select {
+					case resendChan <- struct{}{}:
+					default:
 					}
 				}
 			}
@@ -244,15 +290,6 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp
 				glog.V(0).Infof("master %s asked for the full volume list", masterAddress)
 				vs.store.RequestFullVolumeList()
 			}
-			if in.GetLeader() != "" {
-				current := vs.getCurrentMaster()
-				if !current.Equals(pb.ServerAddress(in.GetLeader())) {
-					glog.V(0).Infof("Volume Server found a new master newLeader: %v instead of %v", in.GetLeader(), current)
-					newLeader = pb.ServerAddress(in.GetLeader())
-					doneChan <- nil
-					return
-				}
-			}
 		}
 	}()

@@ -359,6 +396,11 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp
 				glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
 				return "", err
 			}
+		case <-resendChan:
+			if err = stream.Send(vs.store.CollectHeartbeat()); err != nil {
+				glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterAddress, err)
+				return "", err
+			}
 		case <-volumeTickChan.C:
 			glog.V(4).Infof("volume server %s:%d heartbeat", vs.store.Ip, vs.store.Port)
 			vs.store.MaybeAdjustVolumeMax()
@@ -4,11 +4,17 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"strings"
+	"sync"
 	"testing"
+	"time"

 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/storage"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 )
@@ -60,3 +66,279 @@ func startFakeMasterServerForLeaderLookup(t *testing.T, srv master_pb.SeaweedSer
 	_, port, _ := net.SplitHostPort(lis.Addr().String())
 	return pb.ServerAddress(fmt.Sprintf("127.0.0.1:0.%s", port))
 }
+
+// fakeHeartbeatMaster answers SendHeartbeat the way a master does: a follower
+// names the leader on every heartbeat, the leader answers the first one with
+// its volume options and never names itself.
+type fakeHeartbeatMaster struct {
+	master_pb.UnimplementedSeaweedServer
+	leader     string        // set on a follower
+	release    chan struct{} // a follower waits on it before answering
+	heartbeats chan *master_pb.Heartbeat
+}
+
+func (m *fakeHeartbeatMaster) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServer) error {
+	answered := false
+	for {
+		hb, err := stream.Recv()
+		if err != nil {
+			return err
+		}
+		if m.heartbeats != nil {
+			m.heartbeats <- hb
+		}
+		if m.leader != "" {
+			if m.release != nil {
+				<-m.release
+			}
+			if err := stream.Send(&master_pb.HeartbeatResponse{Leader: m.leader}); err != nil {
+				return err
+			}
+			continue
+		}
+		if !answered {
+			answered = true
+			if err := stream.Send(&master_pb.HeartbeatResponse{VolumeSizeLimit: 1 << 30}); err != nil {
+				return err
+			}
+		}
+	}
+}
+
+func newHeartbeatTestVolumeServer(t *testing.T) *VolumeServer {
+	t.Helper()
+	store := storage.NewStore(nil, "127.0.0.1", 8080, 18080, "http://127.0.0.1:8080", "store-id",
+		[]string{t.TempDir()},
+		[]int32{0},
+		[]util.MinFreeSpace{{}},
+		"",
+		storage.NeedleMapInMemory,
+		[]types.DiskType{types.HardDriveType},
+		nil,
+		3,
+		stats.DefaultDiskIOProbeConfig(),
+	)
+	t.Cleanup(store.Close)
+	vs := &VolumeServer{
+		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
+		store:          store,
+		stopChan:       make(chan bool),
+		isHeartbeating: true,
+	}
+	store.MasterFn = vs.getCurrentMaster
+	return vs
+}
+
+func waitForCurrentMaster(t *testing.T, vs *VolumeServer, want pb.ServerAddress) {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for !vs.getCurrentMaster().Equals(want) {
+		if time.Now().After(deadline) {
+			t.Fatalf("currentMaster = %q, want %q", vs.getCurrentMaster(), want)
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+}
+
+// heartbeatRun is one doHeartbeatWithRetry call running in the background.
+type heartbeatRun struct {
+	done   chan heartbeatResult // receives the call's result once
+	exited chan struct{}
+	stop   func()
+}
+
+type heartbeatResult struct {
+	newLeader pb.ServerAddress
+	err       error
+}
+
+// startHeartbeat runs doHeartbeatWithRetry against master until it returns or
+// is stopped. Its cleanup stops the heartbeat and waits for it, and runs before
+// the cleanups registered earlier, so a failed assertion cannot leave the stream
+// open for the fake master's GracefulStop to wait on.
+func startHeartbeat(t *testing.T, vs *VolumeServer, master pb.ServerAddress) *heartbeatRun {
+	t.Helper()
+	var stopOnce sync.Once
+	run := &heartbeatRun{
+		done:   make(chan heartbeatResult, 1),
+		exited: make(chan struct{}),
+		stop:   func() { stopOnce.Do(func() { close(vs.stopChan) }) },
+	}
+	go func() {
+		defer close(run.exited)
+		newLeader, err := vs.doHeartbeatWithRetry(master, vs.grpcDialOption, time.Hour, 0)
+		run.done <- heartbeatResult{newLeader, err}
+	}()
+	t.Cleanup(func() {
+		run.stop()
+		select {
+		case <-run.exited:
+		case <-time.After(5 * time.Second):
+			t.Error("heartbeat did not stop")
+		}
+	})
+	return run
+}
+
+// A follower must never become the master lookups go to: until it names the
+// leader the previous one stays, and then the named leader takes over before
+// the heartbeat loop has reconnected.
+func TestHeartbeatToFollowerPublishesNamedLeaderOnly(t *testing.T) {
+	const leader = pb.ServerAddress("leader.example.com:9333.19333")
+	follower := &fakeHeartbeatMaster{
+		leader:     string(leader),
+		release:    make(chan struct{}),
+		heartbeats: make(chan *master_pb.Heartbeat, 8),
+	}
+	followerAddr := startFakeMasterServerForLeaderLookup(t, follower)
+	// cleanups run last-in first-out, so a failed assertion still lets the
+	// follower answer before its server stops gracefully
+	var releaseOnce sync.Once
+	release := func() { releaseOnce.Do(func() { close(follower.release) }) }
+	t.Cleanup(release)
+
+	vs := newHeartbeatTestVolumeServer(t)
+	previous := pb.ServerAddress("previous.example.com:9333.19333")
+	vs.setCurrentMaster(previous)
+	run := startHeartbeat(t, vs, followerAddr)
+
+	select {
+	case <-follower.heartbeats:
+	case <-time.After(5 * time.Second):
+		t.Fatal("follower got no heartbeat")
+	}
+	if got := vs.getCurrentMaster(); got != previous {
+		t.Fatalf("currentMaster = %q while connected to follower %q, want unchanged %q", string(got), string(followerAddr), string(previous))
+	}
+	if got := vs.store.MasterFn(); got != previous {
+		t.Fatalf("store master = %q, want %q", got, previous)
+	}
+	release()
+
+	select {
+	case r := <-run.done:
+		if r.err != nil {
+			t.Fatalf("heartbeat: %v", r.err)
+		}
+		if r.newLeader != leader {
+			t.Fatalf("newLeader = %q, want %q", r.newLeader, leader)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("heartbeat did not return after the follower named the leader")
+	}
+	if got := vs.getCurrentMaster(); got != leader {
+		t.Fatalf("currentMaster = %q, want named leader %q", got, leader)
+	}
+}
+
+// The leader's first answer confirms it, and the size limit that answer carries
+// makes the receiver ask for a fresh heartbeat, which the send loop delivers.
+func TestHeartbeatToLeaderPublishesItAndResendsFromSendLoop(t *testing.T) {
+	leader := &fakeHeartbeatMaster{heartbeats: make(chan *master_pb.Heartbeat, 8)}
+	leaderAddr := startFakeMasterServerForLeaderLookup(t, leader)
+
+	vs := newHeartbeatTestVolumeServer(t)
+	run := startHeartbeat(t, vs, leaderAddr)
+
+	waitForCurrentMaster(t, vs, leaderAddr)
+	if got := vs.store.MasterFn(); got != leaderAddr {
+		t.Fatalf("store master = %q, want %q", got, leaderAddr)
+	}
+
+	// the initial volume and ec heartbeats, then the resend
+	for i := 0; i < 3; i++ {
+		select {
+		case <-leader.heartbeats:
+		case <-time.After(5 * time.Second):
+			t.Fatalf("leader got %d heartbeats, want 3", i)
+		}
+	}
+
+	run.stop()
+	select {
+	case r := <-run.done:
+		if r.err != nil {
+			t.Fatalf("heartbeat: %v", r.err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("heartbeat did not stop")
+	}
+}
+
+// duplicateUUIDMaster answers every heartbeat the way a leader holding a stale
+// registration does: a duplicate-uuid response and nothing else.
+type duplicateUUIDMaster struct {
+	master_pb.UnimplementedSeaweedServer
+}
+
+func (m *duplicateUUIDMaster) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServer) error {
+	for {
+		if _, err := stream.Recv(); err != nil {
+			return err
+		}
+		if err := stream.Send(&master_pb.HeartbeatResponse{DuplicatedUuids: []string{"uuid-1"}}); err != nil {
+			return err
+		}
+	}
+}
+
+// A leader that rejects the registration over duplicate uuids still answers
+// lookups, so it is published before the retry error goes out.
+func TestHeartbeatPublishesLeaderBeforeDuplicateUuidError(t *testing.T) {
+	masterAddr := startFakeMasterServerForLeaderLookup(t, &duplicateUUIDMaster{})
+	vs := newHeartbeatTestVolumeServer(t)
+	vs.setCurrentMaster("previous.example.com:9333.19333")
+
+	_, err := vs.doHeartbeatWithRetry(masterAddr, vs.grpcDialOption, time.Hour, 0)
+	if err == nil || !strings.Contains(err.Error(), "duplicate UUIDs") {
+		t.Fatalf("err = %v, want duplicate UUIDs", err)
+	}
+	if got := vs.getCurrentMaster(); got != masterAddr {
+		t.Fatalf("currentMaster = %q, want the answering leader %q", got, masterAddr)
+	}
+}
+
+// After the heartbeat loses its master, lookups move to the leader the other
+// masters report instead of dialing the lost one until a new stream answers.
+func TestRefreshLeaderFromPeersPublishesReportedLeader(t *testing.T) {
+	const leader = pb.ServerAddress("leader.example.com:9333.19333")
+	peer := startFakeMasterServerForLeaderLookup(t, &fakeLeaderConfigServer{leader: string(leader)})
+	vs := &VolumeServer{
+		SeedMasterNodes: []pb.ServerAddress{peer},
+		grpcDialOption:  grpc.WithTransportCredentials(insecure.NewCredentials()),
+		store:           &storage.Store{Ip: "127.0.0.1", Port: 8080},
+	}
+	vs.store.MasterFn = vs.getCurrentMaster
+	vs.setCurrentMaster("lost.example.com:9333.19333")
+
+	vs.refreshLeaderFromPeers(5 * time.Second)
+
+	if got := vs.getCurrentMaster(); got != leader {
+		t.Fatalf("currentMaster = %q, want reported leader %q", string(got), string(leader))
+	}
+	if got := vs.store.MasterFn(); got != leader {
+		t.Fatalf("store master = %q, want %q", string(got), string(leader))
+	}
+}
+
+// Mid-election no master names a leader. The lost master is dead or now a
+// follower that would answer lookups from its own cache, so none is published.
+func TestRefreshLeaderFromPeersClearsLeaderWithoutAnswer(t *testing.T) {
+	peer := startFakeMasterServerForLeaderLookup(t, &fakeLeaderConfigServer{})
+	vs := &VolumeServer{
+		SeedMasterNodes: []pb.ServerAddress{peer},
+		grpcDialOption:  grpc.WithTransportCredentials(insecure.NewCredentials()),
+		store:           &storage.Store{Ip: "127.0.0.1", Port: 8080},
+	}
+	vs.store.MasterFn = vs.getCurrentMaster
+	vs.setCurrentMaster("lost.example.com:9333.19333")
+
+	vs.refreshLeaderFromPeers(5 * time.Second)
+
+	if got := vs.getCurrentMaster(); got != "" {
+		t.Fatalf("currentMaster = %q, want none", string(got))
+	}
+	if got := vs.store.MasterFn(); got != "" {
+		t.Fatalf("store master = %q, want none", string(got))
+	}
+}
@@ -148,6 +148,7 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 	vs.checkWithMaster()

 	vs.store = storage.NewStore(vs.grpcDialOption, ip, port, grpcPort, publicUrl, id, folders, maxCounts, minFreeSpaces, idxFolder, vs.needleMapKind, diskTypes, diskTags, ldbTimeout, diskProbeConfig)
+	vs.store.MasterFn = vs.getCurrentMaster
 	vs.guard = security.NewGuard(whiteList, signingKey, expiresAfterSec, readSigningKey, readExpiresAfterSec)

 	handleStaticResources(adminMux)
@@ -61,14 +61,25 @@ type ReadOption struct {
 	ReadBufferSize int
 }

+// masterAddress returns the master to ask for volume and shard locations, or
+// "" while no leader is known.
+func (s *Store) masterAddress() pb.ServerAddress {
+	if s.MasterFn == nil {
+		return ""
+	}
+	return s.MasterFn()
+}
+
 /*
 * A VolumeServer contains one Store
 */
 type Store struct {
-	MasterAddress       pb.ServerAddress
+	// MasterFn reports the master the volume server currently takes as leader.
+	// The heartbeat loop owns that answer; the store only reads it.
+	MasterFn            func() pb.ServerAddress
 	grpcDialOption      grpc.DialOption
-	volumeSizeLimit     uint64      // read from the master
-	preallocate         atomic.Bool // read from the master
+	volumeSizeLimit     atomic.Uint64 // read from the master
+	preallocate         atomic.Bool   // read from the master
 	Ip                  string
 	Port                int
 	GrpcPort            int
@@ -1302,11 +1313,11 @@ func (s *Store) ConfigureVolume(i needle.VolumeId, replication string) error {
 }

 func (s *Store) SetVolumeSizeLimit(x uint64) {
-	atomic.StoreUint64(&s.volumeSizeLimit, x)
+	s.volumeSizeLimit.Store(x)
 }

 func (s *Store) GetVolumeSizeLimit() uint64 {
-	return atomic.LoadUint64(&s.volumeSizeLimit)
+	return s.volumeSizeLimit.Load()
 }

 func (s *Store) SetPreallocate(x bool) {
@@ -568,7 +568,7 @@ const ecIntervalReadConcurrency = 8

 func (s *Store) readEcShardIntervals(needleId types.NeedleId, ecVolume *erasure_coding.EcVolume, intervals []erasure_coding.Interval) (data []byte, is_deleted bool, err error) {
 	if err = s.cachedLookupEcShardLocations(ecVolume); err != nil {
-		return nil, false, fmt.Errorf("failed to locate shard via master grpc %s: %v", s.MasterAddress, err)
+		return nil, false, fmt.Errorf("failed to locate shard via master grpc %s: %v", s.masterAddress(), err)
 	}

 	var totalSize int
@@ -721,7 +721,7 @@ func (s *Store) cachedLookupEcShardLocations(ecVolume *erasure_coding.EcVolume)

 	glog.V(3).Infof("lookup and cache ec volume %d locations", ecVolume.VolumeId)

-	err = operation.WithMasterServerClient(context.Background(), false, s.MasterAddress, s.grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
+	err = operation.WithMasterServerClient(context.Background(), false, s.masterAddress(), s.grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
 		req := &master_pb.LookupEcVolumeRequest{
 			VolumeId: uint32(ecVolume.VolumeId),
 		}
@@ -78,7 +78,7 @@ func rewindShardLocationsRefresh(ecVolume *erasure_coding.EcVolume, d time.Durat
 func TestCachedLookupEcShardLocationsRefreshesAfterAFailedRead(t *testing.T) {
 	master, masterAddr := startCountingMaster(t)
 	store := &Store{
-		MasterAddress:  masterAddr,
+		MasterFn:       func() pb.ServerAddress { return masterAddr },
 		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
 	}
 	ecVolume := &erasure_coding.EcVolume{
@@ -127,7 +127,7 @@ func TestCachedLookupEcShardLocationsRefreshesAfterAFailedRead(t *testing.T) {
 func TestCachedLookupEcShardLocationsSettlesAfterRefresh(t *testing.T) {
 	master, masterAddr := startCountingMaster(t)
 	store := &Store{
-		MasterAddress:  masterAddr,
+		MasterFn:       func() pb.ServerAddress { return masterAddr },
 		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
 	}
 	ecVolume := &erasure_coding.EcVolume{
@@ -257,7 +257,7 @@ func TestReadOneEcShardIntervalMarksTheMapAfterADirectReadFails(t *testing.T) {
 func TestCachedLookupEcShardLocationsKeepsAMarkRaisedDuringTheLookup(t *testing.T) {
 	master, masterAddr := startCountingMaster(t)
 	store := &Store{
-		MasterAddress:  masterAddr,
+		MasterFn:       func() pb.ServerAddress { return masterAddr },
 		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
 	}
 	ecVolume := &erasure_coding.EcVolume{
@@ -295,7 +295,7 @@ func TestCachedLookupEcShardLocationsKeepsAMarkRaisedDuringTheLookup(t *testing.
 func TestCachedLookupEcShardLocationsKeepsTheMarkWhenTheLookupFails(t *testing.T) {
 	master, masterAddr := startCountingMaster(t)
 	store := &Store{
-		MasterAddress:  masterAddr,
+		MasterFn:       func() pb.ServerAddress { return masterAddr },
 		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
 	}
 	ecVolume := &erasure_coding.EcVolume{
@@ -23,7 +23,7 @@ func (s *Store) ScrubEcVolume(vid needle.VolumeId, mode volume_server_pb.VolumeS
 		return 0, nil, []error{fmt.Errorf("EC volume id %d not found", vid)}
 	}
 	if err := s.cachedLookupEcShardLocations(ecv); err != nil {
-		return 0, nil, []error{fmt.Errorf("failed to locate shard via master grpc %s: %v", s.MasterAddress, err)}
+		return 0, nil, []error{fmt.Errorf("failed to locate shard via master grpc %s: %v", s.masterAddress(), err)}
 	}
 	return s.scrubEcVolumeWalk(ecv, mode, forceDeletedNeedlesCheck)
 }
@@ -56,7 +56,7 @@ func (s *Store) ScrubEcVolumeMerged(merged *erasure_coding.MergedEcRuntimes, mod
 		}
 	}
 	if err := s.cachedLookupEcShardLocations(ecv); err != nil {
-		return 0, nil, []error{fmt.Errorf("failed to locate shard via master grpc %s: %v", s.MasterAddress, err)}
+		return 0, nil, []error{fmt.Errorf("failed to locate shard via master grpc %s: %v", s.masterAddress(), err)}
 	}

 	files, shardInfos, errs := s.scrubEcVolumeWalk(ecv, mode, forceDeletedNeedlesCheck)