filer: announce to master only once gRPC serves, and leave before it stops (#11700)

* wdclient: gate master registration until the server is serving

The first KeepConnected send registers the client into the master's
cluster list, which is what routes S3 and peer traffic to it. A filer
connects to the master long before its gRPC port serves (peer discovery,
store bootstrap), so it announced itself while still refusing
connections. announceCh defers only the registration send, letting the
stream machinery connect early without joining the list; ListMasters
exposes the configured masters for callers that cannot wait on the
stream.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer: announce to the master only once gRPC serves

NewFilerServer previously registered on the master's filer list during
startup, before the gRPC listener was serving — and on a fresh store
before peer bootstrap finished — so a restarting filer could be routed
to while refusing connections. The announce channel is closed right
after grpcS.Serve starts. ListExistingPeerUpdates now iterates the
configured masters directly since GetMaster would block on the gated
stream.

On shutdown, cancelling the master context ends the KeepConnected
stream during the leave phase, so the master drops the filer node
before gRPC stops accepting; fs.Shutdown cancels it as a fallback.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* wdclient: test that the announce gate delays registration

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* wdclient: unblock queries behind the announce gate

The gate holds the registration send, but setting currentMaster only after
the first response left GetMaster blocked for the whole gated window. Any
startup path that queries the master before serving (e.g. loading a
chunked filer.conf) deadlocked the constructor, so the gate never opened.

Mark the master queryable once the stream is established; registration and
leader redirect still wait on the gate and first response as before.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer: keep the fullest peer list from any master

ListExistingPeerUpdates returned the first non-nil answer, so a follower
master could seed the peer list with a stale partial view before the
leader replied. Keep the longest response instead.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer: keep master queries usable after the shutdown disconnect

Leaving the filer list early ends the KeepConnected stream, which cleared
currentMaster and left the final metadata-log flush unable to reach the
master during drain. Keep the last serving address queryable after a
canceled disconnect, and move the Shutdown fallback cancel behind the
filer's own shutdown so callers that skip LeaveLockRing flush against a
live stream.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* cluster: propagate a bounded context through ListExistingPeerUpdates

The helper dialed and queried masters on context.Background, so one
unresponsive master could stall the whole peer scan and ignore caller
cancellation. Pass ctx through, bound each master query in the filer's
scan, and stop when the caller cancels.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer: wait for cluster removal before draining gRPC

Cancelling the KeepConnected stream only signals removal; the master
drops the filer when its stream handler unwinds. Until then new requests
can still route to this filer while its listener is closing. Poll the
configured masters, bounded, until none still lists this filer.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* wdclient: serialize master discovery access

ListMasters made peer scans a second concurrent accessor of mc.masters
alongside tryAllMasters, while ServerDiscovery refreshes and reads its
address list unsynchronized. Guard the refresh and every instance copy
behind one lock.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* cluster, filer: prefer the leader's peer list and report lookup errors

Keeping the longest answer let a stale ex-leader's list win: after a
leader change the old master retains registrations until its streams
end, so a fresh filer could bootstrap from an unreachable peer. Ask the
masters for the raft leader and take its list; only when the leader is
unreachable fall back to the fullest answer.

ListExistingPeerUpdates also returned an empty slice on RPC failure,
which a fresh filer read as "no peers" and skipped bootstrap, serving
without the existing files. Propagate the error and retry the scan
during startup, before the announce gate opens.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer: bound the removal wait and retry unconfirmed masters

Each master query got its own two-second timeout while the five-second
deadline was only checked between scans, so three stalled queries could
stretch the wait past its budget. A failed query also counted as
confirmed removal, letting gRPC drain while a master still listed the
filer. Derive every query context from one overall deadline, track which
masters confirmed absence, and retry the rest until they answer or the
budget expires.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* wdclient: run the SRV lookup outside the masters lock

ListMasters and tryAllMasters held mastersLock across the DNS lookup, so
a slow resolver stalled GetMasters and ListMasterSet readers past their
request deadlines. Split the lookup from the list swap: resolve the SRV
record unlocked, then lock only to save it.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* wdclient: bound WaitUntilConnected in the announce-gate test

The wait ran on the caller context with no deadline, so a regression
that left currentMaster unset hung the whole test run instead of
failing the check. Give it the same three-second budget as the
assertions that follow.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* wdclient: order SRV refreshes and reuse the cached list on shutdown

* filer: confirm an empty leader peer list before accepting it

* filer: announce only after the startup peers' replay catches up

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/cluster/master_client.go

@@ -10,10 +10,13 @@ import (
 	"google.golang.org/grpc"
 )
 
-func ListExistingPeerUpdates(master pb.ServerAddress, grpcDialOption grpc.DialOption, filerGroup string, clientType string) (existingNodes []*master_pb.ClusterNodeUpdate) {
+// ListExistingPeerUpdates lists the cluster nodes one master currently tracks.
+// An empty result is only a successful empty answer; a failed RPC returns the
+// error so callers do not mistake "unknown" for "absent".
+func ListExistingPeerUpdates(ctx context.Context, master pb.ServerAddress, grpcDialOption grpc.DialOption, filerGroup string, clientType string) (existingNodes []*master_pb.ClusterNodeUpdate, err error) {
 
-	if grpcErr := pb.WithMasterClient(context.Background(), false, master, grpcDialOption, false, func(client master_pb.SeaweedClient) error {
-		resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
+	err = pb.WithMasterClient(ctx, false, master, grpcDialOption, false, func(client master_pb.SeaweedClient) error {
+		resp, err := client.ListClusterNodes(ctx, &master_pb.ListClusterNodesRequest{
 			ClientType: clientType,
 			FilerGroup: filerGroup,
 		})
@@ -31,8 +34,24 @@ func ListExistingPeerUpdates(master pb.ServerAddress, grpcDialOption grpc.DialOp
 			})
 		}
 		return nil
-	}); grpcErr != nil {
-		glog.V(0).Infof("connect to %s: %v", master, grpcErr)
+	})
+	if err != nil {
+		glog.V(0).Infof("connect to %s: %v", master, err)
 	}
 	return
 }
+
+// LookupClusterLeader asks one master for the raft leader it currently sees.
+// Any reachable master answers, including followers; it returns "" while the
+// cluster is mid-election.
+func LookupClusterLeader(ctx context.Context, master pb.ServerAddress, grpcDialOption grpc.DialOption) (leader pb.ServerAddress, err error) {
+	err = pb.WithMasterClient(ctx, false, master, grpcDialOption, false, func(client master_pb.SeaweedClient) error {
+		resp, err := client.GetMasterConfiguration(ctx, &master_pb.GetMasterConfigurationRequest{})
+		if err != nil {
+			return err
+		}
+		leader = pb.ServerAddress(resp.Leader)
+		return nil
+	})
+	return
+}

weed/command/filer.go

@@ -382,8 +382,10 @@ func (fo *FilerOptions) startFiler() {
 		}
 	}
 
+	announce := make(chan struct{})
 	fs, nfs_err := weed_server.NewFilerServer(defaultMux, publicVolumeMux, &weed_server.FilerOption{
 		Masters:                   fo.masters,
+		AnnounceCh:                announce,
 		FilerGroup:                *fo.filerGroup,
 		Collection:                *fo.collection,
 		DefaultReplication:        *fo.defaultReplicaPlacement,
@@ -500,6 +502,13 @@ func (fo *FilerOptions) startFiler() {
 	go grpcS.Serve(grpcL)
 	pb.ServeGrpcOnLocalSocket(grpcS, grpcPort)
 
+	// Membership in the master's filer list routes S3 and peer traffic here;
+	// announce once the gRPC port is serving and the metadata replay has
+	// caught up with the peers found at startup, so the filer does not take
+	// lock-ring ownership while a peer's older events can still land on top
+	// of its fresh writes. A cap bounds the wait.
+	go fs.AnnounceOnceCaughtUp(announce, time.Now())
+
 	// Helper to gracefully stop the gRPC server, waiting for active RPCs.
 	gracefulTimeout := fo.gracefulStopTimeout
 	if gracefulTimeout <= 0 {

weed/filer/filer.go

@@ -254,8 +254,96 @@ func (f *Filer) AggregateFromPeers(self pb.ServerAddress, existingNodes []*maste
 
 }
 
-func (f *Filer) ListExistingPeerUpdates(ctx context.Context) (existingNodes []*master_pb.ClusterNodeUpdate) {
-	return cluster.ListExistingPeerUpdates(f.GetMaster(ctx), f.GrpcDialOption, f.MasterClient.FilerGroup, cluster.FilerType)
+// peerLookupTimeout bounds each master query so one unresponsive master
+// cannot stall the peer scan.
+const peerLookupTimeout = 10 * time.Second
+
+// leaderSettleTimeout bounds how long an empty peer list from the leader is
+// re-polled: a freshly elected leader has not yet received the filers'
+// reconnects, so an immediate empty answer can just mean membership has not
+// migrated yet. A persistent empty answer is then checked against the other
+// masters below before it is accepted as an empty cluster.
+const leaderSettleTimeout = 10 * time.Second
+const leaderSettleInterval = time.Second
+
+// ListExistingPeerUpdates returns the peer list to bootstrap from. The
+// leader's list is authoritative: after a leader change an old master keeps
+// its registrations until its streams end, so the longest answer is not
+// necessarily the current one. The leader's answer wins whenever it can be
+// found and queried; only when the leader is unreachable does the scan fall
+// back to the fullest answer from the other masters. It returns an error when
+// no master answers, so a failed lookup is not mistaken for an empty cluster.
+func (f *Filer) ListExistingPeerUpdates(ctx context.Context) (existingNodes []*master_pb.ClusterNodeUpdate, err error) {
+	masters := f.MasterClient.ListMasters()
+	var queryErrs []error
+
+	var leader pb.ServerAddress
+	for _, master := range masters {
+		if ctx.Err() != nil {
+			break
+		}
+		lookupCtx, cancel := context.WithTimeout(ctx, peerLookupTimeout)
+		l, lookErr := cluster.LookupClusterLeader(lookupCtx, master, f.GrpcDialOption)
+		cancel()
+		if lookErr != nil {
+			queryErrs = append(queryErrs, lookErr)
+			continue
+		}
+		if l != "" {
+			leader = l
+			break
+		}
+	}
+	if leader != "" {
+		settleDeadline := time.Now().Add(leaderSettleTimeout)
+		for {
+			lookupCtx, cancel := context.WithTimeout(ctx, peerLookupTimeout)
+			nodes, listErr := cluster.ListExistingPeerUpdates(lookupCtx, leader, f.GrpcDialOption, f.MasterClient.FilerGroup, cluster.FilerType)
+			cancel()
+			if listErr != nil {
+				queryErrs = append(queryErrs, fmt.Errorf("list peers from leader %s: %w", leader, listErr))
+				break
+			}
+			if len(nodes) > 0 {
+				return nodes, nil
+			}
+			if ctx.Err() != nil || !time.Now().Before(settleDeadline) {
+				break
+			}
+			select {
+			case <-ctx.Done():
+			case <-time.After(leaderSettleInterval):
+			}
+		}
+	}
+
+	answered := false
+	for _, master := range masters {
+		if ctx.Err() != nil {
+			break
+		}
+		lookupCtx, cancel := context.WithTimeout(ctx, peerLookupTimeout)
+		nodes, listErr := cluster.ListExistingPeerUpdates(lookupCtx, master, f.GrpcDialOption, f.MasterClient.FilerGroup, cluster.FilerType)
+		cancel()
+		if listErr != nil {
+			queryErrs = append(queryErrs, listErr)
+			continue
+		}
+		answered = true
+		if len(nodes) > len(existingNodes) {
+			existingNodes = nodes
+		}
+	}
+	if !answered {
+		if ctxErr := ctx.Err(); ctxErr != nil {
+			return nil, ctxErr
+		}
+		if err = errors.Join(queryErrs...); err == nil {
+			err = fmt.Errorf("no masters configured")
+		}
+		return nil, fmt.Errorf("list existing peers: %w", err)
+	}
+	return existingNodes, nil
 }
 
 func (f *Filer) SetStore(store FilerStore) (isFresh bool) {

weed/filer/meta_aggregator.go

@@ -273,6 +273,21 @@ func lowWatermarkOf(watermarks map[pb.ServerAddress]int64) int64 {
 	return low
 }
 
+// PeersCaughtUpTo reports whether every listed peer's stream has delivered
+// an event or idle heartbeat at or past tsNs, i.e. no pre-tsNs backlog can
+// still arrive from it. A peer no longer tracked counts as caught up: it
+// cannot deliver a backlog anymore.
+func (ma *MetaAggregator) PeersCaughtUpTo(peers []pb.ServerAddress, tsNs int64) bool {
+	ma.peerWatermarksLock.Lock()
+	defer ma.peerWatermarksLock.Unlock()
+	for _, peer := range peers {
+		if ts, found := ma.peerWatermarks[peer]; found && ts < tsNs {
+			return false
+		}
+	}
+	return true
+}
+
 func (ma *MetaAggregator) HasRemotePeers() bool {
 	ma.peerChansLock.Lock()
 	defer ma.peerChansLock.Unlock()

weed/mq/broker/broker_server.go

@@ -125,7 +125,10 @@ func NewMessageBroker(option *MessageQueueBrokerOption, grpcDialOption grpc.Dial
 	)
 	glog.V(0).Info("Started idle partition cleanup task (check: 1m, timeout: 5m)")
 
-	existingNodes := cluster.ListExistingPeerUpdates(mqBroker.MasterClient.GetMaster(context.Background()), grpcDialOption, option.FilerGroup, cluster.FilerType)
+	existingNodes, listErr := cluster.ListExistingPeerUpdates(context.Background(), mqBroker.MasterClient.GetMaster(context.Background()), grpcDialOption, option.FilerGroup, cluster.FilerType)
+	if listErr != nil {
+		glog.Warningf("broker failed to list existing peers: %v", listErr)
+	}
 	for _, newNode := range existingNodes {
 		mqBroker.OnBrokerUpdate(newNode, time.Now())
 	}

weed/pb/server_discovery.go

@@ -24,8 +24,19 @@ func NewServiceDiscoveryFromMap(m map[string]ServerAddress) (sd *ServerDiscovery
 // RefreshBySrvIfAvailable performs a DNS SRV lookup and updates list with the results
 // of the lookup
 func (sd *ServerDiscovery) RefreshBySrvIfAvailable() {
+	if newList := sd.LookupSrvInstances(); newList != nil {
+		sd.SetInstances(newList)
+	}
+}
+
+// LookupSrvInstances resolves the SRV record without touching the stored
+// list. It returns nil when there is no SRV record, the lookup fails, or the
+// result has no well-formed names. Running it without holding a lock keeps a
+// slow resolver from blocking GetInstances callers; callers serialize the
+// SetInstances that saves the result.
+func (sd *ServerDiscovery) LookupSrvInstances() []ServerAddress {
 	if sd.srvRecord == nil {
-		return
+		return nil
 	}
 	newList, err := sd.srvRecord.LookUp()
 	if err != nil {
@@ -33,8 +44,14 @@ func (sd *ServerDiscovery) RefreshBySrvIfAvailable() {
 	}
 	if newList == nil || len(newList) == 0 {
 		glog.V(0).Infof("looked up SRV for %s, but found no well-formed names", *sd.srvRecord)
-		return
+		return nil
 	}
+	return newList
+}
+
+// SetInstances saves a resolved address list. Callers racing a refresh must
+// serialize this with their GetInstances reads.
+func (sd *ServerDiscovery) SetInstances(newList []ServerAddress) {
 	if !reflect.DeepEqual(sd.list, newList) {
 		sd.list = newList
 	}

weed/server/filer_server.go

@@ -91,6 +91,10 @@ type FilerOption struct {
 	TusSessionExpiry          time.Duration
 	S3ConfigFile              string // optional path to static S3 identity config file
 	CredentialManager         *credential.CredentialManager
+	// AnnounceCh, when set, must be closed before the filer registers on the
+	// master; it joins the filer list only once the gRPC port is serving and
+	// the metadata replay has caught up with the peers found at startup.
+	AnnounceCh chan struct{}
 	// AllowUntrustedRemoteEndpoints lets a read of a remote-only entry dial a
 	// mounted endpoint that resolves to a loopback / private / metadata host.
 	AllowUntrustedRemoteEndpoints bool
@@ -155,6 +159,16 @@ type FilerServer struct {
 	ringPeerIPs      atomic.Pointer[ringPeerIPs]
 	ringResolveGroup singleflight.Group
 
+	// masterCtx drives KeepConnectedToMaster; cancelling it ends the stream
+	// so the master drops this filer from its list before gRPC stops.
+	masterCtx    context.Context
+	masterCancel context.CancelFunc
+
+	// startupPeers are the filers discovered during NewFilerServer, before
+	// this filer announced itself; AnnounceOnceCaughtUp waits for their
+	// replayed backlogs before the filer joins the master's list.
+	startupPeers []pb.ServerAddress
+
 	// entryLockTable serializes mutations to the same entry path on this filer.
 	// CreateEntry takes it today; UpdateEntry and DeleteEntry are intended to take
 	// it too as their callers route a key's writes to this node, making it the
@@ -272,7 +286,11 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 	fs.checkWithMaster()
 
 	go stats.LoopPushingMetric("filer", string(fs.option.Host), fs.metricsAddress, fs.metricsIntervalSec)
-	go fs.filer.MasterClient.KeepConnectedToMaster(context.Background())
+	if option.AnnounceCh != nil {
+		fs.filer.MasterClient.SetAnnounceCh(option.AnnounceCh)
+	}
+	fs.masterCtx, fs.masterCancel = context.WithCancel(context.Background())
+	go fs.filer.MasterClient.KeepConnectedToMaster(fs.masterCtx)
 
 	fs.option.recursiveDelete = v.GetBool("filer.options.recursive_delete")
 	v.SetDefault("filer.options.buckets_folder", "/buckets")
@@ -321,7 +339,20 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 		readonlyMux.HandleFunc("/", fs.filerGuard.WhiteList(requestIDMiddleware(fs.readonlyFilerHandler)))
 	}
 
-	existingNodes := fs.filer.ListExistingPeerUpdates(context.Background())
+	// A failed peer scan is not an empty cluster: a fresh filer that skipped
+	// bootstrap here would serve without the existing files, so retry until a
+	// master answers. This runs before the announce gate opens, so the filer
+	// is not yet advertised.
+	var existingNodes []*master_pb.ClusterNodeUpdate
+	for {
+		var listErr error
+		existingNodes, listErr = fs.filer.ListExistingPeerUpdates(context.Background())
+		if listErr == nil {
+			break
+		}
+		glog.Warningf("%s cannot list existing peers: %v; retrying", option.Host, listErr)
+		time.Sleep(2 * time.Second)
+	}
 	startFromTime := time.Now().Add(-filer.LogFlushInterval)
 	if isFresh {
 		glog.V(0).Infof("%s bootstrap from peers %+v", option.Host, existingNodes)
@@ -335,6 +366,15 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 	}
 	fs.filer.AggregateFromPeers(option.Host, existingNodes, startFromTime)
 
+	// A peer replaying its backlog applies the old owner's events over this
+	// filer's fresh writes if it takes lock-ring ownership before the replay
+	// finishes, so the announce gate waits for these peers to catch up.
+	for _, node := range existingNodes {
+		if addr := pb.ServerAddress(node.Address); !addr.Equals(option.Host) {
+			fs.startupPeers = append(fs.startupPeers, addr)
+		}
+	}
+
 	fs.filer.LoadFilerConf()
 
 	fs.filer.LoadRemoteStorageConfAndMapping()
@@ -390,6 +430,12 @@ func (fs *FilerServer) Shutdown() {
 		fs.remoteCacheEvictCancel()
 	}
 	fs.filer.Shutdown()
+	// LeaveLockRing already ended the master stream in the normal path; this
+	// covers callers that skip it, and runs after the filer's own shutdown so
+	// the final metadata-log flush can still reach the master.
+	if fs.masterCancel != nil {
+		fs.masterCancel()
+	}
 }
 
 // priorOwnerWindowSkew covers peers and gateways that start their prior-owner
@@ -417,6 +463,106 @@ func (fs *FilerServer) leaveLockRingWithin(removalTimeout, budget time.Duration)
 	case <-time.After(budget):
 		glog.Warningf("LockRing: %s did not finish leaving within %v, shutting down anyway", fs.option.Host, budget)
 	}
+	// End the master stream so the master drops this filer from the cluster
+	// list before gRPC stops accepting connections.
+	if fs.masterCancel != nil {
+		fs.masterCancel()
+	}
+	fs.waitForClusterRemoval()
+}
+
+// announceCatchUpCap bounds how long startup holds the master registration
+// while the metadata replay catches up with the peers found at startup. Past
+// it the filer announces anyway, as it did when the gate opened on serve.
+const announceCatchUpCap = time.Minute
+
+// AnnounceOnceCaughtUp closes announce once every peer found at startup has
+// replayed its backlog. Each peer's stream advances a watermark per event or
+// idle heartbeat, so a watermark past waitStart means nothing older can still
+// land on top of this filer's fresh writes. Called once gRPC is serving, so
+// routed traffic is answerable the moment the filer is listed.
+func (fs *FilerServer) AnnounceOnceCaughtUp(announce chan struct{}, waitStart time.Time) {
+	defer close(announce)
+	if len(fs.startupPeers) == 0 || fs.filer.MetaAggregator == nil {
+		return
+	}
+	deadline := time.Now().Add(announceCatchUpCap)
+	for {
+		// Take the channel before reading the watermarks so a rise in
+		// between still wakes this wait (see DeliveryWatermarkAdvancedChan).
+		advanced := fs.filer.MetaAggregator.DeliveryWatermarkAdvancedChan()
+		if fs.filer.MetaAggregator.PeersCaughtUpTo(fs.startupPeers, waitStart.UnixNano()) {
+			glog.V(0).Infof("%s caught up with %d startup peers, announcing to masters", fs.option.Host, len(fs.startupPeers))
+			return
+		}
+		remaining := time.Until(deadline)
+		if remaining <= 0 {
+			glog.Warningf("%s metadata replay not caught up after %v, announcing anyway", fs.option.Host, announceCatchUpCap)
+			return
+		}
+		select {
+		case <-advanced:
+		case <-time.After(min(remaining, time.Second)):
+		}
+	}
+}
+
+// masterRemovalWaitBudget bounds how long shutdown waits for masters to
+// drop this filer before gRPC drains anyway.
+const masterRemovalWaitBudget = 5 * time.Second
+
+// waitForClusterRemoval polls the configured masters until none still lists
+// this filer. The master removes membership when the KeepConnected handler
+// unwinds, which is asynchronous to cancel, so new requests could otherwise
+// route to this filer after its gRPC listener has already closed.
+func (fs *FilerServer) waitForClusterRemoval() {
+	ctx, cancel := context.WithTimeout(context.Background(), masterRemovalWaitBudget)
+	defer cancel()
+	confirmedAbsent := make(map[pb.ServerAddress]struct{})
+	for ctx.Err() == nil {
+		pending := false
+		// The cached list keeps a slow SRV lookup from stretching the
+		// removal-wait budget.
+		for _, master := range fs.filer.MasterClient.ListCachedMasters() {
+			if _, ok := confirmedAbsent[master]; ok {
+				continue
+			}
+			if ctx.Err() != nil {
+				return
+			}
+			// The per-query timeout shares the overall deadline, so a stalled
+			// master cannot stretch the wait past the budget.
+			queryCtx, queryCancel := context.WithTimeout(ctx, 2*time.Second)
+			nodes, err := cluster.ListExistingPeerUpdates(queryCtx, master, fs.grpcDialOption, fs.filer.MasterClient.FilerGroup, cluster.FilerType)
+			queryCancel()
+			if err != nil {
+				// A failed query is not confirmed removal; retry until the
+				// master answers or the overall deadline expires.
+				pending = true
+				continue
+			}
+			listed := false
+			for _, node := range nodes {
+				if node.Address == string(fs.option.Host) {
+					listed = true
+					break
+				}
+			}
+			if listed {
+				pending = true
+			} else {
+				confirmedAbsent[master] = struct{}{}
+			}
+		}
+		if !pending {
+			return
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(200 * time.Millisecond):
+		}
+	}
 }
 
 func (fs *FilerServer) leaveLockRing(removalTimeout time.Duration) {

weed/wdclient/filer_client.go

@@ -341,7 +341,11 @@ func (fc *FilerClient) refreshFilerList() {
 	generation := fc.peerUpdateGeneration()
 
 	// Query master for filers in our group
-	updates := cluster.ListExistingPeerUpdates(currentMaster, fc.grpcDialOption, fc.filerGroup, cluster.FilerType)
+	updates, err := cluster.ListExistingPeerUpdates(context.Background(), currentMaster, fc.grpcDialOption, fc.filerGroup, cluster.FilerType)
+	if err != nil {
+		glog.V(1).Infof("FilerClient: failed to list filers in group '%s': %v", fc.filerGroup, err)
+		return
+	}
 
 	if len(updates) == 0 {
 		glog.V(2).Infof("FilerClient: no filers found in group '%s'", fc.filerGroup)

weed/wdclient/masterclient.go

@@ -145,14 +145,18 @@ func (p *masterVolumeProvider) LookupVolumeIds(ctx context.Context, volumeIds []
 type MasterClient struct {
 	*vidMapClient // Embedded cache with shared logic
 
-	FilerGroup           string
-	clientType           string
-	clientHost           pb.ServerAddress
-	rack                 string
-	currentMaster        pb.ServerAddress
-	lastServedMaster     pb.ServerAddress
-	currentMasterLock    sync.RWMutex
-	masters              pb.ServerDiscovery
+	FilerGroup        string
+	clientType        string
+	clientHost        pb.ServerAddress
+	rack              string
+	currentMaster     pb.ServerAddress
+	lastServedMaster  pb.ServerAddress
+	currentMasterLock sync.RWMutex
+	masters           pb.ServerDiscovery
+	mastersLock       sync.Mutex
+	// refreshLock serializes SRV refreshes so an older lookup finishing
+	// second cannot overwrite a newer saved list.
+	refreshLock          sync.Mutex
 	grpcDialOption       grpc.DialOption
 	grpcTimeout          time.Duration // Timeout for gRPC calls to master
 	OnPeerUpdate         func(update *master_pb.ClusterNodeUpdate, startFrom time.Time)
@@ -167,6 +171,55 @@ type MasterClient struct {
 	streamLock      sync.Mutex
 	stream          master_pb.Seaweed_KeepConnectedClient
 	leavingLockRing bool
+
+	// announceCh, when non-nil, gates the registration send that adds this
+	// client to the master's list, so a server can connect and query masters
+	// before it is ready to be routed to.
+	announceCh chan struct{}
+}
+
+// SetAnnounceCh sets a channel that must be closed before this client
+// registers itself on the master.
+func (mc *MasterClient) SetAnnounceCh(ch chan struct{}) {
+	mc.announceCh = ch
+}
+
+// ListMasters returns the configured master addresses without waiting for a
+// connection, for callers that run before KeepConnectedToMaster is connected.
+func (mc *MasterClient) ListMasters() []pb.ServerAddress {
+	mc.refreshMasters()
+	return mc.masterInstances()
+}
+
+// ListCachedMasters returns the most recently resolved master addresses
+// without a fresh SRV lookup, for callers on a deadline that cannot wait for
+// DNS.
+func (mc *MasterClient) ListCachedMasters() []pb.ServerAddress {
+	return mc.masterInstances()
+}
+
+// refreshMasters resolves the SRV record outside mastersLock so a slow DNS
+// lookup does not stall masterInstances readers; the lock only covers saving
+// the resolved list. refreshLock is held across lookup and save so
+// overlapping refreshes keep resolution order.
+func (mc *MasterClient) refreshMasters() {
+	mc.refreshLock.Lock()
+	defer mc.refreshLock.Unlock()
+	newList := mc.masters.LookupSrvInstances()
+	if newList == nil {
+		return
+	}
+	mc.mastersLock.Lock()
+	defer mc.mastersLock.Unlock()
+	mc.masters.SetInstances(newList)
+}
+
+// masterInstances returns a consistent copy of the configured masters,
+// serialized against SRV refreshes.
+func (mc *MasterClient) masterInstances() []pb.ServerAddress {
+	mc.mastersLock.Lock()
+	defer mc.mastersLock.Unlock()
+	return mc.masters.GetInstances()
 }
 
 func NewMasterClient(grpcDialOption grpc.DialOption, filerGroup string, clientType string, clientHost pb.ServerAddress, clientDataCenter string, rack string, masters pb.ServerDiscovery) *MasterClient {
@@ -210,8 +263,9 @@ func (mc *MasterClient) SetOnMasterChangeFn(fn func(previous, current pb.ServerA
 func (mc *MasterClient) tryAllMasters(ctx context.Context) {
 	var nextHintedLeader pb.ServerAddress
 	failedMasters := make(map[pb.ServerAddress]struct{})
-	mc.masters.RefreshBySrvIfAvailable()
-	for _, master := range mc.masters.GetInstances() {
+	mc.refreshMasters()
+	masterAddrs := mc.masterInstances()
+	for _, master := range masterAddrs {
 		if _, failed := failedMasters[master]; failed {
 			continue
 		}
@@ -234,7 +288,12 @@ func (mc *MasterClient) tryAllMasters(ctx context.Context) {
 				}
 			}
 		}
-		mc.setCurrentMaster("")
+		// After a canceled disconnect the last master still answers unary
+		// queries (e.g. the metadata-log flush during filer shutdown), so
+		// only clear the serving address on an unintended disconnect.
+		if ctx.Err() == nil {
+			mc.setCurrentMaster("")
+		}
 	}
 }
 
@@ -254,6 +313,19 @@ func (mc *MasterClient) tryConnectToMaster(ctx context.Context, master pb.Server
 		}
 		glog.V(1).Infof("%s.%s masterClient gRPC stream established to %s in %v", mc.FilerGroup, mc.clientType, master, time.Since(connectStartTime))
 
+		// The stream is usable for queries before registration, so callers
+		// blocked in GetMaster (e.g. loading a chunked filer.conf during
+		// startup) proceed while the announce gate still holds.
+		mc.setCurrentMaster(master)
+
+		if mc.announceCh != nil {
+			select {
+			case <-mc.announceCh:
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		}
+
 		mc.streamLock.Lock()
 		err = stream.Send(&master_pb.KeepConnectedRequest{
 			FilerGroup:    mc.FilerGroup,
@@ -588,14 +660,14 @@ func (mc *MasterClient) GetMaster(ctx context.Context) pb.ServerAddress {
 // See GetMaster() for important initialization contract details.
 func (mc *MasterClient) GetMasters(ctx context.Context) []pb.ServerAddress {
 	mc.WaitUntilConnected(ctx)
-	return mc.masters.GetInstances()
+	return mc.masterInstances()
 }
 
 // ListMasterSet returns a set of configured master addresses keyed by their
 // canonical http form. Unlike GetMasters this does not wait for a connection,
 // so it is safe to call from admission paths that must stay non-blocking.
 func (mc *MasterClient) ListMasterSet() map[string]struct{} {
-	addrs := mc.masters.GetInstances()
+	addrs := mc.masterInstances()
 	set := make(map[string]struct{}, len(addrs))
 	for _, a := range addrs {
 		set[a.ToHttpAddress()] = struct{}{}
@@ -659,7 +731,7 @@ func (mc *MasterClient) KeepConnectedToMaster(ctx context.Context) {
 }
 
 func (mc *MasterClient) FindLeaderFromOtherPeers(myMasterAddress pb.ServerAddress) (leader string) {
-	for _, master := range mc.masters.GetInstances() {
+	for _, master := range mc.masterInstances() {
 		if master == myMasterAddress {
 			continue
 		}

weed/wdclient/masterclient_announce_test.go

@@ -0,0 +1,64 @@
+package wdclient
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+// A filer connects to the master before its gRPC port serves so it can query
+// masters during startup; the registration send that adds it to the master's
+// filer list must wait until it is actually serving.
+func TestAnnounceGateDelaysRegistration(t *testing.T) {
+	srv := &fakeKeepConnectedServer{requests: make(chan *master_pb.KeepConnectedRequest, 4)}
+	addr := startFakeMasterServer(t, srv)
+	mc := NewMasterClient(
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		"", "filer", "filer1:18888", "", "",
+		*pb.NewServiceDiscoveryFromMap(map[string]pb.ServerAddress{"m": addr}),
+	)
+	gate := make(chan struct{})
+	mc.SetAnnounceCh(gate)
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	go mc.KeepConnectedToMaster(ctx)
+
+	// WaitUntilConnected returns once the stream is established, before any
+	// registration: queries (e.g. loading a chunked filer.conf during
+	// startup) must not block behind the gate. Bound the wait so a
+	// regression fails fast instead of hanging the test run.
+	waitCtx, waitCancel := context.WithTimeout(ctx, 3*time.Second)
+	defer waitCancel()
+	mc.WaitUntilConnected(waitCtx)
+	if err := waitCtx.Err(); err != nil {
+		t.Fatalf("master stream not established within 3s: %v", err)
+	}
+	got := make(chan pb.ServerAddress, 1)
+	go func() { got <- mc.GetMaster(ctx) }()
+	select {
+	case m := <-got:
+		if m != addr {
+			t.Fatalf("GetMaster = %q, want %q", m, addr)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("GetMaster blocked behind the announce gate")
+	}
+
+	// Registration must not have been sent yet.
+	select {
+	case req := <-srv.requests:
+		t.Fatalf("client registered before the gate opened: %+v", req)
+	default:
+	}
+
+	close(gate)
+	req := nextRequest(t, srv.requests)
+	if req.ClientAddress != "filer1:18888" {
+		t.Fatalf("registration ClientAddress = %q, want filer1:18888", req.ClientAddress)
+	}
+}