From 27fa285d48482d8930ed753d6de1302aacee27ad Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 23 Oct 2026 02:04:27 -0700 Subject: [PATCH] filer: do not stall startup waiting out the leader-settle window (#11972) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 | 27 +++++---- weed/pb/filer_pb/filer_client_bfs.go | 6 +- weed/server/filer_server.go | 83 ++++++++++++++++++++++++++-- 3 files changed, 97 insertions(+), 19 deletions(-) diff --git a/weed/filer/filer.go b/weed/filer/filer.go index 522df042a9..5a0090e61b 100644 --- a/weed/filer/filer.go +++ b/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() diff --git a/weed/pb/filer_pb/filer_client_bfs.go b/weed/pb/filer_pb/filer_client_bfs.go index e4ec78d675..05c2a4387b 100644 --- a/weed/pb/filer_pb/filer_client_bfs.go +++ b/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 } } diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index 41c7256172..8e09191723 100644 --- a/weed/server/filer_server.go +++ b/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