filer: do not stall startup waiting out the leader-settle window (#11972)

* filer: do not stall startup waiting out the leader-settle window

ListExistingPeerUpdates re-polled the leader for up to 10s when it
reported zero filers. A fresh single-filer cluster can never gain peers,
so every filer start paid the full settle timeout before listening —
past the 6s health window in weed mini and short probe budgets.

Take the leader's first answer instead; the fallback fullest-answer
scan below still covers a leader that is mid-convergence.

* filer: bound leader-settle re-poll below the startup health window

* filer: update ListExistingPeerUpdates doc for empty-leader fallback

* filer: re-check empty membership at the announce gate before registering

A fresh filer that scans empty membership during a master restart or
re-election skipped bootstrap and announced immediately, serving missing
files until a peer happened to reconnect — and a late peer only got an
incremental subscription, so pre-existing metadata was never backfilled.

The announce gate now re-polls ListExistingPeerUpdates while the scan can
still be disproved (10s), backfills from any late peers before their
streams start, and then waits for their catch-up like startup peers. A
genuinely empty cluster pays a delayed master listing rather than a
stalled startup, since this runs after the ports already serve.

* filer: overlap the late-peer copy cutoff and stream start

A peer event stamped just behind this filer's clock could land after
the traversal read but carry a timestamp below the stream start —
the startup scan margins both cutoffs by LogFlushInterval for exactly
this skew; mirror it for peers discovered at the announce gate.

* filer: bound the late-peer bootstrap copy by the shared announce deadline

MaybeBootstrapFromOnePeer now takes the caller's context so a stalled
peer metadata copy cannot out-wait the announce cap, and the bootstrap
re-check shares one announceCatchUpCap deadline with the catch-up wait
instead of each spending a full minute.

* filer: merge bootstrap entries by commit time instead of blind insert

The late-peer copy runs while gRPC already serves, so a blind InsertEntry
could replace a write the local filer just accepted. Routing each entry
through applyTraversedEntry orders it against the local commit time under
the replay locks — a fresher local write survives the backfill.

---------

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

weed/filer/filer.go

@@ -190,7 +190,7 @@ func (f *Filer) buildRemoteStorageClient(ctx context.Context, remoteConf *remote
 	return remote_storage.GetRemoteStorage(remoteConf)
 }
 
-func (f *Filer) MaybeBootstrapFromOnePeer(self pb.ServerAddress, existingNodes []*master_pb.ClusterNodeUpdate, snapshotTime time.Time) (err error) {
+func (f *Filer) MaybeBootstrapFromOnePeer(ctx context.Context, self pb.ServerAddress, existingNodes []*master_pb.ClusterNodeUpdate, snapshotTime time.Time) (err error) {
 	if len(existingNodes) == 0 {
 		return
 	}
@@ -205,8 +205,13 @@ func (f *Filer) MaybeBootstrapFromOnePeer(self pb.ServerAddress, existingNodes [
 	glog.V(0).Infof("bootstrap from %v clientId:%d", earliestNode.Address, f.UniqueFilerId)
 
 	return pb.WithFilerClient(false, f.UniqueFilerId, pb.ServerAddress(earliestNode.Address), f.GrpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
-		return filer_pb.StreamBfs(client, "/", snapshotTime.UnixNano(), func(parentPath util.FullPath, entry *filer_pb.Entry) error {
-			return f.Store.InsertEntry(context.Background(), FromPbEntry(string(parentPath), entry))
+		return filer_pb.StreamBfs(ctx, client, "/", snapshotTime.UnixNano(), func(parentPath util.FullPath, pbEntry *filer_pb.Entry, entryTsNs int64) error {
+			entry := FromPbEntry(string(parentPath), pbEntry)
+			peerTsNs := entryTsNs
+			if peerTsNs == 0 {
+				peerTsNs = entry.Attr.Mtime.UnixNano()
+			}
+			return f.applyTraversedEntry(ctx, entry, peerTsNs)
 		})
 	})
 
@@ -271,19 +276,19 @@ func (f *Filer) AggregateFromPeers(self pb.ServerAddress, existingNodes []*maste
 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
+// re-polled: a freshly elected leader may not yet have the filers' reconnects,
+// so an immediate empty answer can just mean membership has not migrated yet.
+// It must stay below the health window a fresh filer gets to start serving
+// (weed mini allows 6s) — a filerless cluster pays the whole window per start.
+const leaderSettleTimeout = 3 * 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
+// necessarily the current one. A non-empty answer from the leader wins; a
+// leader that is unreachable or reports an empty list falls 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()

weed/pb/filer_pb/filer_client_bfs.go

@@ -115,9 +115,9 @@ func processOneDirectory(ctx context.Context, filerClient FilerClient, parentPat
 
 }
 
-func StreamBfs(client SeaweedFilerClient, dir util.FullPath, olderThanTsNs int64, fn func(parentPath util.FullPath, entry *Entry) error) (err error) {
+func StreamBfs(ctx context.Context, client SeaweedFilerClient, dir util.FullPath, olderThanTsNs int64, fn func(parentPath util.FullPath, entry *Entry, entryTsNs int64) error) (err error) {
 	glog.V(0).Infof("TraverseBfsMetadata %v if before %v", dir, time.Unix(0, olderThanTsNs))
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	stream, err := client.TraverseBfsMetadata(ctx, &TraverseBfsMetadataRequest{
 		Directory: string(dir),
@@ -133,7 +133,7 @@ func StreamBfs(client SeaweedFilerClient, dir util.FullPath, olderThanTsNs int64
 			}
 			return fmt.Errorf("traverse bfs metadata: %w", err)
 		}
-		if err := fn(util.FullPath(resp.Directory), resp.Entry); err != nil {
+		if err := fn(util.FullPath(resp.Directory), resp.Entry, resp.EntryTsNs); err != nil {
 			return err
 		}
 	}

weed/server/filer_server.go

@@ -176,6 +176,12 @@ type FilerServer struct {
 	// replayed backlogs before the filer joins the master's list.
 	startupPeers []pb.ServerAddress
 
+	// peerBootstrapPending marks a fresh store whose startup peer scan came
+	// back empty: an empty answer during a master restart can just mean the
+	// existing filers have not reconnected yet, so AnnounceOnceCaughtUp
+	// re-checks membership before registering.
+	peerBootstrapPending bool
+
 	// 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
@@ -367,9 +373,12 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 	startFromTime := time.Now().Add(-filer.LogFlushInterval)
 	if isFresh {
 		glog.V(0).Infof("%s bootstrap from peers %+v", option.Host, existingNodes)
-		if err := fs.filer.MaybeBootstrapFromOnePeer(option.Host, existingNodes, startFromTime); err != nil {
+		if err := fs.filer.MaybeBootstrapFromOnePeer(context.Background(), option.Host, existingNodes, startFromTime); err != nil {
 			glog.Fatalf("%s bootstrap from %+v: %v", option.Host, existingNodes, err)
 		}
+		// An empty scan can be a settling master rather than an empty cluster;
+		// the announce gate then re-checks membership before registering.
+		fs.peerBootstrapPending = len(existingNodes) == 0
 	}
 	v.SetDefault("filer.options.s3.empty_folder_cleanup_delay", "2m")
 	if d, err := time.ParseDuration(v.GetString("filer.options.s3.empty_folder_cleanup_delay")); err == nil {
@@ -494,16 +503,26 @@ const announceCatchUpCap = time.Minute
 // 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 {
+	ma := fs.filer.MetaAggregator
+	if ma == nil {
 		return
 	}
+	// The bootstrap re-check and the catch-up wait share one deadline so a
+	// late peer cannot stretch the registration hold past the cap.
 	deadline := time.Now().Add(announceCatchUpCap)
+	peers := fs.startupPeers
+	if fs.peerBootstrapPending {
+		peers = append(peers, fs.awaitPeerBootstrap(deadline)...)
+	}
+	if len(peers) == 0 {
+		return
+	}
 	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))
+		advanced := ma.DeliveryWatermarkAdvancedChan()
+		if ma.PeersCaughtUpTo(peers, waitStart.UnixNano()) {
+			glog.V(0).Infof("%s caught up with %d startup peers, announcing to masters", fs.option.Host, len(peers))
 			return
 		}
 		remaining := time.Until(deadline)
@@ -518,6 +537,60 @@ func (fs *FilerServer) AnnounceOnceCaughtUp(announce chan struct{}, waitStart ti
 	}
 }
 
+// emptyMembershipConfirmTimeout bounds the announce hold for a peer to
+// disprove an empty startup scan on a fresh store. This runs after the ports
+// already serve, so a genuinely empty cluster pays a delayed master listing,
+// not a stalled startup.
+const emptyMembershipConfirmTimeout = 10 * time.Second
+
+// awaitPeerBootstrap re-polls membership while a fresh store's empty startup
+// scan can still be disproved: masters restarted or re-elected report no
+// filers until the existing ones reconnect. Peers found late are backfilled
+// from before their streams start, then reported so the caller waits for
+// their catch-up like startup peers. The whole wait — including each peer
+// metadata copy — is bounded by the shared announce deadline.
+func (fs *FilerServer) awaitPeerBootstrap(deadline time.Time) (peers []pb.ServerAddress) {
+	confirmBy := time.Now().Add(emptyMembershipConfirmTimeout)
+	for time.Now().Before(deadline) {
+		ctx, cancel := context.WithDeadline(context.Background(), deadline)
+		nodes, listErr := fs.filer.ListExistingPeerUpdates(ctx)
+		cancel()
+		if listErr != nil {
+			glog.Warningf("%s cannot re-check existing peers: %v; announcing anyway", fs.option.Host, listErr)
+			return nil
+		}
+		var remotes []*master_pb.ClusterNodeUpdate
+		for _, node := range nodes {
+			if !pb.ServerAddress(node.Address).Equals(fs.option.Host) {
+				remotes = append(remotes, node)
+			}
+		}
+		if len(remotes) == 0 {
+			if !time.Now().Before(confirmBy) {
+				return nil
+			}
+			time.Sleep(time.Second)
+			continue
+		}
+		bootstrapFrom := time.Now().Add(-filer.LogFlushInterval)
+		ctx, cancel = context.WithDeadline(context.Background(), deadline)
+		err := fs.filer.MaybeBootstrapFromOnePeer(ctx, fs.option.Host, remotes, bootstrapFrom)
+		cancel()
+		if err != nil {
+			glog.Warningf("%s bootstrap from late peers %+v: %v; retrying", fs.option.Host, remotes, err)
+			time.Sleep(2 * time.Second)
+			continue
+		}
+		glog.V(0).Infof("%s bootstrap from late-arriving peers %+v", fs.option.Host, remotes)
+		for _, node := range remotes {
+			peers = append(peers, pb.ServerAddress(node.Address))
+			fs.filer.MetaAggregator.OnPeerUpdate(node, bootstrapFrom)
+		}
+		return peers
+	}
+	return nil
+}
+
 // masterRemovalWaitBudget bounds how long shutdown waits for masters to
 // drop this filer before gRPC drains anyway.
 const masterRemovalWaitBudget = 5 * time.Second