From da1d7444b81ed1bf270a1a1132731705aef0b65e Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 25 Oct 2026 06:17:42 -0700 Subject: [PATCH] master: shorten the first-write hold after a fresh topology id (#12059) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * master: release the post-mint allocation hold after a quiet pulse DeferAllocation armed a fixed warmup window whenever the leader minted a fresh topology id, so the first write on a brand-new cluster blocked for the whole window even though no prior state could exist. The window still has to cover a rebuilt master's veteran servers, which report on their first post-reconnect heartbeat. Track the last heartbeat that could carry prior cluster state — a server joining, held volumes or shards, a nonzero id bound, or a foreign topology id — and lift the hold one pulse after it goes quiet, capped by the original window. * master: make the post-mint allocation hold configurable master.volume_growth.allocation_deferral overrides the warmup window a leader holds volume id allocation for after minting a fresh topology id. Unset keeps the existing window; "0s" disables the hold for setups that know the cluster is fresh (test harnesses, single-node labs); any other duration replaces the cap while the quiet-pulse early release still applies. * master: re-check the allocation deadline while it is held NextVolumeId slept for the whole computed delay, so a heartbeat landing mid-sleep that pushed the quiet deadline out was ignored — the pending allocation could still run ahead of a straggler's report. Wait in short slices, re-evaluating the deadline and the growth freeze each round. * master: count only ids above the issued bound as prior-state evidence Heartbeat echoes of volumes and shards this master allocated during the hold refreshed the quiet window, so each growth allocation re-extended the deferral until the hard cap. Evidence now requires an id beyond the bound this leader has issued, a higher reported bound, or a foreign topology id; a joining server (dn == nil) still counts once. * master: reject an unparseable allocation_deferral instead of disabling the hold Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * master: do not count a skipped heartbeat as a joining server Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test: assert the allocation hold extends past the original quiet deadline Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * master: release the post-mint hold after two quiet pulses One pulse proved short for a veteran that reconnects on a staggered restart: with only five seconds of quiet, allocation could resume and reuse ids before a slower server landed its first heartbeat. Two pulses still halves the fresh-cluster first-write stall versus the full warmup cap, and known-fresh setups keep the "0s" override. * master: pick once more when assign times out as growth concludes The pick loop's final attempt can predate the growth it triggered concluding: growth registers its volumes in the layout before clearing the pending flag, so a timeout landing on that boundary used to shed a hard 406 over volumes that had just arrived. Pick once more after the loop in both the http and grpc assign paths before deciding how to shed. --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/command/scaffold/master.toml | 5 ++ weed/server/master_grpc_server.go | 52 +++++++++++++++++ weed/server/master_grpc_server_assign.go | 53 ++++++++++------- .../master_grpc_server_prior_state_test.go | 53 +++++++++++++++++ weed/server/master_server.go | 7 +++ weed/server/master_server_handlers.go | 10 ++++ weed/server/raft_common.go | 2 +- weed/topology/growth_freeze_test.go | 47 +++++++++++++++ weed/topology/topology.go | 57 ++++++++++++++++--- 9 files changed, 257 insertions(+), 29 deletions(-) create mode 100644 weed/server/master_grpc_server_prior_state_test.go diff --git a/weed/command/scaffold/master.toml b/weed/command/scaffold/master.toml index dfa8472030..fd16616b1b 100644 --- a/weed/command/scaffold/master.toml +++ b/weed/command/scaffold/master.toml @@ -63,6 +63,11 @@ copy_other = 1 # create n x 1 = n actual volumes threshold = 0.9 # create threshold reservation_timeout = "5m"# capacity reservation timeout before unreleased reservations expire disable = false # disables volume growth if true +# allocation_deferral = "15s" # hold the first volume id allocation after a fresh + # topology id is minted, giving a rebuilt master's + # veteran volume servers time to report. The hold + # lifts early once heartbeats stay clean for two + # pulses. "0s" disables it (e.g. test setups). # configuration flags for replication [master.replication] diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go index dcba381e6d..92a2e35161 100644 --- a/weed/server/master_grpc_server.go +++ b/weed/server/master_grpc_server.go @@ -103,6 +103,55 @@ func announceVolume(message *master_pb.VolumeLocation, vid uint32, isRemote, isR } } +// reportsPriorClusterState tells whether a heartbeat could come from a server +// still carrying cluster state from before this leader's term: it holds a +// volume or shard id above the bound this master has issued, has seen a +// higher volume id bound, or last synced with a different topology id. +// Echoes of this term's own allocations stay under the bound and do not count. +func reportsPriorClusterState(heartbeat *master_pb.Heartbeat, topologyId string, bound needle.VolumeId) bool { + if needle.VolumeId(heartbeat.MaxVolumeIdSeen) > bound || + (heartbeat.LastSeenTopologyId != "" && heartbeat.LastSeenTopologyId != topologyId) { + return true + } + unissued := func(id uint32) bool { return needle.VolumeId(id) > bound } + for _, v := range heartbeat.Volumes { + if unissued(v.Id) { + return true + } + } + for _, v := range heartbeat.NewVolumes { + if unissued(v.Id) { + return true + } + } + for _, v := range heartbeat.ChangedVolumes { + if unissued(v.Id) { + return true + } + } + for _, v := range heartbeat.DeletedVolumes { + if unissued(v.Id) { + return true + } + } + for _, s := range heartbeat.EcShards { + if unissued(s.Id) { + return true + } + } + for _, s := range heartbeat.NewEcShards { + if unissued(s.Id) { + return true + } + } + for _, s := range heartbeat.DeletedEcShards { + if unissued(s.Id) { + return true + } + } + return false +} + func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServer) error { var dn *topology.DataNode @@ -182,6 +231,9 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ continue } + if (dn == nil && heartbeat.Ip != "") || reportsPriorClusterState(heartbeat, ms.Topo.GetTopologyId(), ms.Topo.GetMaxVolumeId()) { + ms.Topo.NoteAllocationEvidence() + } ms.Topo.Sequence.SetMax(heartbeat.MaxFileKey) ms.Topo.UpAdjustMaxVolumeId(needle.VolumeId(heartbeat.MaxVolumeIdSeen)) if len(heartbeat.Volumes) > 0 { diff --git a/weed/server/master_grpc_server_assign.go b/weed/server/master_grpc_server_assign.go index 703a6ab3a9..3a07d35ee8 100644 --- a/weed/server/master_grpc_server_assign.go +++ b/weed/server/master_grpc_server_assign.go @@ -199,27 +199,15 @@ func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest if dn == nil { continue } - var replicas []*master_pb.Location - for _, r := range dnList.Rest() { - replicas = append(replicas, &master_pb.Location{ - Url: r.Url(), - PublicUrl: r.PublicUrl, - GrpcPort: uint32(r.GrpcPort), - DataCenter: r.GetDataCenterId(), - }) + return assignResponse(fid, count, dnList, ms.guard), nil + } + // The last in-loop pick can predate the growth it triggered concluding: + // growth registers its volumes before clearing the pending flag, so pick + // once more before shedding. + if fid, count, dnList, _, err := ms.Topo.PickForWrite(req.Count, option, vl, req.ExpectedDataSize); err == nil { + if dnList.Head() != nil { + return assignResponse(fid, count, dnList, ms.guard), nil } - return &master_pb.AssignResponse{ - Fid: fid, - Location: &master_pb.Location{ - Url: dn.Url(), - PublicUrl: dn.PublicUrl, - GrpcPort: uint32(dn.GrpcPort), - DataCenter: dn.GetDataCenterId(), - }, - Count: count, - Auth: string(security.GenJwtForVolumeServer(ms.guard.SigningKey(), ms.guard.ExpiresAfterSec(), fid)), - Replicas: replicas, - }, nil } // The initiator timed out with its growth still pending: shed retryably // rather than failing the write that growth is about to satisfy. @@ -235,6 +223,31 @@ func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest return nil, lastErr } +func assignResponse(fid string, count uint64, dnList *topology.VolumeLocationList, guard *security.Guard) *master_pb.AssignResponse { + var replicas []*master_pb.Location + for _, r := range dnList.Rest() { + replicas = append(replicas, &master_pb.Location{ + Url: r.Url(), + PublicUrl: r.PublicUrl, + GrpcPort: uint32(r.GrpcPort), + DataCenter: r.GetDataCenterId(), + }) + } + dn := dnList.Head() + return &master_pb.AssignResponse{ + Fid: fid, + Location: &master_pb.Location{ + Url: dn.Url(), + PublicUrl: dn.PublicUrl, + GrpcPort: uint32(dn.GrpcPort), + DataCenter: dn.GetDataCenterId(), + }, + Count: count, + Auth: string(security.GenJwtForVolumeServer(guard.SigningKey(), guard.ExpiresAfterSec(), fid)), + Replicas: replicas, + } +} + // describeDiskLayout names the disk layout an assign targets. The empty disk // type is the legacy unlabeled layout on servers that were never started with // -disk; naming it "hdd" sends operators looking for servers that were never diff --git a/weed/server/master_grpc_server_prior_state_test.go b/weed/server/master_grpc_server_prior_state_test.go new file mode 100644 index 0000000000..21e53272dc --- /dev/null +++ b/weed/server/master_grpc_server_prior_state_test.go @@ -0,0 +1,53 @@ +package weed_server + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/stretchr/testify/assert" +) + +func TestReportsPriorClusterState(t *testing.T) { + topoId := "topo-1" + + // a quiet heartbeat from a synced server is not evidence + assert.False(t, reportsPriorClusterState(&master_pb.Heartbeat{ + LastSeenTopologyId: topoId, + }, topoId, 5)) + + // echoes of volume ids this master issued are not evidence + assert.False(t, reportsPriorClusterState(&master_pb.Heartbeat{ + MaxVolumeIdSeen: 5, + NewVolumes: []*master_pb.VolumeShortInformationMessage{{Id: 5}}, + ChangedVolumes: []*master_pb.VolumeInformationMessage{{Id: 3}}, + LastSeenTopologyId: topoId, + }, topoId, 5)) + + // ids above the issued bound come from before this term + assert.True(t, reportsPriorClusterState(&master_pb.Heartbeat{ + MaxVolumeIdSeen: 6, + }, topoId, 5)) + assert.True(t, reportsPriorClusterState(&master_pb.Heartbeat{ + Volumes: []*master_pb.VolumeInformationMessage{{Id: 6}}, + }, topoId, 5)) + assert.True(t, reportsPriorClusterState(&master_pb.Heartbeat{ + NewVolumes: []*master_pb.VolumeShortInformationMessage{{Id: 6}}, + }, topoId, 5)) + assert.True(t, reportsPriorClusterState(&master_pb.Heartbeat{ + DeletedVolumes: []*master_pb.VolumeShortInformationMessage{{Id: 6}}, + }, topoId, 5)) + assert.True(t, reportsPriorClusterState(&master_pb.Heartbeat{ + EcShards: []*master_pb.VolumeEcShardInformationMessage{{Id: 42}}, + }, topoId, 5)) + assert.True(t, reportsPriorClusterState(&master_pb.Heartbeat{ + NewEcShards: []*master_pb.VolumeEcShardInformationMessage{{Id: 42}}, + }, topoId, 5)) + assert.True(t, reportsPriorClusterState(&master_pb.Heartbeat{ + DeletedEcShards: []*master_pb.VolumeEcShardInformationMessage{{Id: 42}}, + }, topoId, 5)) + + // a heartbeat last synced under another topology id is prior state + assert.True(t, reportsPriorClusterState(&master_pb.Heartbeat{ + LastSeenTopologyId: "topo-0", + }, topoId, 5)) +} diff --git a/weed/server/master_server.go b/weed/server/master_server.go index 595259c7b1..0fe35ccf2e 100644 --- a/weed/server/master_server.go +++ b/weed/server/master_server.go @@ -175,6 +175,13 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se ms.Topo.SetGrowthFreezeConfig( v.GetBool("master.volume_growth.freeze_on_topology_mismatch"), v.GetDuration("master.volume_growth.state_loss_recovery_wait")) + if str := strings.TrimSpace(v.GetString("master.volume_growth.allocation_deferral")); str != "" { + if d, err := time.ParseDuration(str); err != nil { + glog.Warningf("invalid master.volume_growth.allocation_deferral %q: %v", str, err) + } else { + ms.Topo.SetAllocationDeferral(d) + } + } ms.Topo.QuarantineVolumeFn = func(dn *topology.DataNode, vid needle.VolumeId, readOnly bool) error { err := operation.WithVolumeServerClient(false, dn.ServerAddress(), ms.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) diff --git a/weed/server/master_server_handlers.go b/weed/server/master_server_handlers.go index 4bb583dc7b..315a81104e 100644 --- a/weed/server/master_server_handlers.go +++ b/weed/server/master_server_handlers.go @@ -261,6 +261,16 @@ func (ms *MasterServer) dirAssignHandler(w http.ResponseWriter, r *http.Request) } } + // The last in-loop pick can predate the growth it triggered concluding: + // growth registers its volumes before clearing the pending flag, so pick + // once more before shedding. + if fid, count, dnList, _, err := ms.Topo.PickForWrite(requestedCount, option, vl, expectedDataSize); err == nil { + if dn := dnList.Head(); dn != nil { + ms.maybeAddJwtAuthorization(w, fid, true) + writeJsonQuiet(w, r, http.StatusOK, operation.AssignResult{Fid: fid, Url: dn.Url(), PublicUrl: dn.PublicUrl, Count: count}) + return + } + } // See Assign: initiator that timed out with growth still pending stays retryable. if initiatedGrow && vl.HasGrowRequest() && ms.Topo.AvailableSpaceFor(option) > 0 { w.Header().Set("Retry-After", "1") diff --git a/weed/server/raft_common.go b/weed/server/raft_common.go index 2d13124698..50dd647254 100644 --- a/weed/server/raft_common.go +++ b/weed/server/raft_common.go @@ -65,7 +65,7 @@ func EnsureTopologyId(topo *topology.Topology, checkLeaderFn func() bool, persis } else { glog.V(0).Infof("TopologyId generated: %s", topologyId) } - topo.DeferAllocation(topo.WarmupDuration()) + topo.DeferAllocation(topo.AllocationDeferralDuration()) break } } diff --git a/weed/topology/growth_freeze_test.go b/weed/topology/growth_freeze_test.go index 3b296b14df..f838010bd9 100644 --- a/weed/topology/growth_freeze_test.go +++ b/weed/topology/growth_freeze_test.go @@ -144,3 +144,50 @@ func TestAllocationDelay(t *testing.T) { t.Fatalf("expired deferral should not delay: %v", wait) } } + +func TestAllocationDelayEarlyRelease(t *testing.T) { + topo := NewTopology("topo", nil, 128, 1, false) + + topo.DeferAllocation(time.Minute) + wait := topo.allocationDelay() + if wait <= 0 || wait > 2*time.Second { + t.Fatalf("quiet window should bound the deferral near two pulses, got %v", wait) + } + + time.Sleep(2100 * time.Millisecond) + if wait := topo.allocationDelay(); wait > 0 { + t.Fatalf("two quiet pulses should release the deferral early, got %v", wait) + } +} + +func TestAllocationDelayExtendedByEvidence(t *testing.T) { + topo := NewTopology("topo", nil, 128, 1, false) + + topo.DeferAllocation(time.Minute) + time.Sleep(1100 * time.Millisecond) + topo.NoteAllocationEvidence() + time.Sleep(1200 * time.Millisecond) + if wait := topo.allocationDelay(); wait <= 0 { + t.Fatal("evidence should extend the deferral past the original quiet deadline") + } +} + +func TestAllocationDeferralDuration(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + + if d := topo.AllocationDeferralDuration(); d != topo.WarmupDuration() { + t.Fatalf("unset deferral should default to the warmup window, got %v", d) + } + topo.SetAllocationDeferral(0) + if d := topo.AllocationDeferralDuration(); d != 0 { + t.Fatalf("zero deferral should disable the hold, got %v", d) + } + topo.SetAllocationDeferral(2 * time.Second) + if d := topo.AllocationDeferralDuration(); d != 2*time.Second { + t.Fatalf("configured deferral should win, got %v", d) + } + topo.SetAllocationDeferral(-1) + if d := topo.AllocationDeferralDuration(); d != topo.WarmupDuration() { + t.Fatalf("negative deferral should restore the default, got %v", d) + } +} diff --git a/weed/topology/topology.go b/weed/topology/topology.go index 3ceda5de78..017f3c8eb1 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -92,12 +92,14 @@ type Topology struct { lastPersistedTopologyId atomic.Value lastPersistedGrowthFrozen atomic.Bool - growthFrozen atomic.Bool - growthFrozenAt atomic.Int64 - freezeOnMismatch atomic.Bool - stateLossRecoveryWait atomic.Int64 - allocationNotBefore atomic.Int64 - topologyInitPending atomic.Bool + growthFrozen atomic.Bool + growthFrozenAt atomic.Int64 + freezeOnMismatch atomic.Bool + stateLossRecoveryWait atomic.Int64 + allocationNotBefore atomic.Int64 + allocationEvidenceSince atomic.Int64 + allocationDeferral atomic.Int64 + topologyInitPending atomic.Bool reportedTopologyIdsLock sync.Mutex pendingReportedTopologyIds map[string]struct{} @@ -324,6 +326,8 @@ func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, puls t.chanFullVolumes = make(chan storage.VolumeInfo) t.chanCrowdedVolumes = make(chan storage.VolumeInfo) + t.allocationDeferral.Store(-1) + t.Configuration = &Configuration{} t.dataNodeIndex = make(map[string]*DataNode) t.quarantinePushLocks = make(map[quarantinedVolume]*quarantinePushLock) @@ -621,6 +625,11 @@ func (t *Topology) NextVolumeId() (needle.VolumeId, error) { } if wait := t.allocationDelay(); wait > 0 { glog.V(0).Infof("deferring volume id allocation for %v", wait) + } + for wait := t.allocationDelay(); wait > 0; wait = t.allocationDelay() { + if wait > 500*time.Millisecond { + wait = 500 * time.Millisecond + } time.Sleep(wait) if t.IsGrowthFrozen() { return 0, fmt.Errorf("volume growth is frozen after master state loss; verify the cluster inventory, then run: volume.growth.resume") @@ -1577,11 +1586,43 @@ func (t *Topology) SetTopologyInitPending(pending bool) { // mints a fresh topology id: on a rebuilt master that window gives veteran // servers a pulse to raise the bound or trip the freeze. func (t *Topology) DeferAllocation(d time.Duration) { - t.allocationNotBefore.Store(time.Now().Add(d).UnixNano()) + now := time.Now() + t.allocationNotBefore.Store(now.Add(d).UnixNano()) + t.allocationEvidenceSince.Store(now.UnixNano()) +} + +// NoteAllocationEvidence records a heartbeat signal that could still carry +// prior cluster state — a server joining or reporting held volumes, a raised +// id bound, or a foreign topology id. The deferral lifts two pulses after the +// last such signal, so a cluster with nothing left to report stops waiting. +func (t *Topology) NoteAllocationEvidence() { + t.allocationEvidenceSince.Store(time.Now().UnixNano()) +} + +// SetAllocationDeferral overrides the post-mint allocation hold: zero disables +// it, a positive duration sets the cap, negative restores the warmup window. +func (t *Topology) SetAllocationDeferral(d time.Duration) { + t.allocationDeferral.Store(int64(d)) +} + +// AllocationDeferralDuration returns the configured post-mint hold, or one +// warmup window when unset. +func (t *Topology) AllocationDeferralDuration() time.Duration { + if d := t.allocationDeferral.Load(); d >= 0 { + return time.Duration(d) + } + return t.WarmupDuration() } func (t *Topology) allocationDelay() time.Duration { - return time.Until(time.Unix(0, t.allocationNotBefore.Load())) + deadline := t.allocationNotBefore.Load() + if deadline == 0 { + return 0 + } + if quietUntil := t.allocationEvidenceSince.Load() + 2*int64(t.pulse)*int64(time.Second); quietUntil < deadline { + deadline = quietUntil + } + return time.Until(time.Unix(0, deadline)) } // SetMaxVolumeIdFile mirrors the allocation bound to a file outside raft, so