master: replicate the growth freeze and gate it on veteran heartbeats (#11763)

weed/server/master_grpc_server.go

@@ -184,7 +184,9 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 
 		ms.Topo.Sequence.SetMax(heartbeat.MaxFileKey)
 		ms.Topo.UpAdjustMaxVolumeId(needle.VolumeId(heartbeat.MaxVolumeIdSeen))
-		ms.Topo.CheckReportedTopologyId(heartbeat.LastSeenTopologyId)
+		if len(heartbeat.Volumes) > 0 {
+			ms.Topo.CheckReportedTopologyId(heartbeat.LastSeenTopologyId)
+		}
 		if dn == nil {
 			// Skip delta heartbeat for volume server versions  better than 3.28 https://github.com/seaweedfs/seaweedfs/pull/3630
 			if heartbeat.Ip == "" {

weed/server/raft_server.go

@@ -60,6 +60,10 @@ func (s StateMachine) Save() ([]byte, error) {
 		MaxVolumeId: s.topo.GetMaxVolumeId(),
 		TopologyId:  s.topo.GetTopologyId(),
 	}
+	if s.topo.IsGrowthFrozen() {
+		state.GrowthFrozen = new(bool)
+		*state.GrowthFrozen = true
+	}
 	glog.V(1).Infof("Save raft state %+v", state)
 	return json.Marshal(state)
 }
@@ -79,6 +83,9 @@ func (s StateMachine) Recovery(data []byte) error {
 	if err := s.topo.PersistMaxVolumeId(state.MaxVolumeId); err != nil {
 		glog.Warningf("persist max volume id %d: %v", state.MaxVolumeId, err)
 	}
+	if state.GrowthFrozen != nil {
+		s.topo.SetGrowthFrozen(*state.GrowthFrozen)
+	}
 	return nil
 }
 
@@ -101,16 +108,24 @@ func (s *StateMachine) Apply(l *hashicorpRaft.Log) interface{} {
 	if err := s.topo.PersistMaxVolumeId(state.MaxVolumeId); err != nil {
 		glog.Warningf("persist max volume id %d: %v", state.MaxVolumeId, err)
 	}
+	if state.GrowthFrozen != nil {
+		s.topo.SetGrowthFrozen(*state.GrowthFrozen)
+	}
 
 	glog.V(1).Infoln("max volume id", before, "==>", s.topo.GetMaxVolumeId())
 	return nil
 }
 
 func (s *StateMachine) Snapshot() (hashicorpRaft.FSMSnapshot, error) {
-	return &topology.MaxVolumeIdCommand{
+	state := &topology.MaxVolumeIdCommand{
 		MaxVolumeId: s.topo.GetMaxVolumeId(),
 		TopologyId:  s.topo.GetTopologyId(),
-	}, nil
+	}
+	if s.topo.IsGrowthFrozen() {
+		state.GrowthFrozen = new(bool)
+		*state.GrowthFrozen = true
+	}
+	return state, nil
 }
 
 func (s *StateMachine) Restore(r io.ReadCloser) error {

weed/topology/cluster_commands.go

@@ -13,6 +13,9 @@ import (
 type MaxVolumeIdCommand struct {
 	MaxVolumeId needle.VolumeId `json:"maxVolumeId"`
 	TopologyId  string          `json:"topologyId"`
+	// GrowthFrozen replicates the volume-growth freeze through raft so a
+	// leader change does not silently lift it. nil means no change.
+	GrowthFrozen *bool `json:"growthFrozen,omitempty"`
 }
 
 func NewMaxVolumeIdCommand(value needle.VolumeId, topologyId string) *MaxVolumeIdCommand {
@@ -46,6 +49,9 @@ func (c *MaxVolumeIdCommand) Apply(server raft.Server) (interface{}, error) {
 	if err := topo.PersistMaxVolumeId(c.MaxVolumeId); err != nil {
 		glog.Warningf("persist max volume id %d: %v", c.MaxVolumeId, err)
 	}
+	if c.GrowthFrozen != nil {
+		topo.SetGrowthFrozen(*c.GrowthFrozen)
+	}
 	glog.V(1).Infoln("max volume id", before, "==>", topo.GetMaxVolumeId())
 
 	return nil, nil

weed/topology/growth_freeze_test.go

@@ -68,6 +68,49 @@ func TestCheckReportedTopologyIdBeforeTopologyIdKnown(t *testing.T) {
 	}
 }
 
+func TestGrowthFreezeSurvivesMirrorRestore(t *testing.T) {
+	dir := t.TempDir()
+	file := dir + "/max_volume_id"
+
+	frozen := NewTopology("topo", nil, 128, 3, false)
+	frozen.SetGrowthFreezeConfig(true, 0)
+	frozen.SetMaxVolumeIdFile(file)
+	frozen.SetTopologyId("cluster-a")
+	frozen.CheckReportedTopologyId("cluster-b")
+	if !frozen.IsGrowthFrozen() {
+		t.Fatal("expected frozen growth")
+	}
+	if err := frozen.PersistMaxVolumeId(42); err != nil {
+		t.Fatalf("persist: %v", err)
+	}
+
+	restarted := NewTopology("topo", nil, 128, 3, false)
+	restarted.SetGrowthFreezeConfig(true, 0)
+	restarted.SetMaxVolumeIdFile(file)
+	if !restarted.IsGrowthFrozen() {
+		t.Fatal("freeze flag should survive a mirror restore")
+	}
+	if !restarted.ResumeVolumeGrowth() {
+		t.Fatal("restored freeze should be resumable")
+	}
+}
+
+func TestGrowthFreezeCommandApply(t *testing.T) {
+	topo := NewTopology("topo", nil, 128, 3, false)
+	topo.SetGrowthFreezeConfig(true, 0)
+
+	frozen := true
+	topo.SetGrowthFrozen(frozen)
+	if !topo.IsGrowthFrozen() {
+		t.Fatal("replicated freeze should set the flag")
+	}
+	unfrozen := false
+	topo.SetGrowthFrozen(unfrozen)
+	if topo.IsGrowthFrozen() {
+		t.Fatal("replicated unfreeze should clear the flag")
+	}
+}
+
 func TestAllocationDelay(t *testing.T) {
 	topo := NewTopology("topo", nil, 128, 3, false)
 

weed/topology/topology.go

@@ -406,6 +406,9 @@ func (t *Topology) NextVolumeId() (needle.VolumeId, error) {
 	if wait := t.allocationDelay(); wait > 0 {
 		glog.V(0).Infof("deferring volume id allocation for %v", wait)
 		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")
+		}
 	}
 	vid := t.GetMaxVolumeId()
 	next := vid.Next()
@@ -980,10 +983,11 @@ func (t *Topology) CheckReportedTopologyId(reported string) {
 	if own == "" || reported == own {
 		return
 	}
+	t.growthFrozenAt.Store(time.Now().UnixNano())
 	if t.growthFrozen.CompareAndSwap(false, true) {
-		t.growthFrozenAt.Store(time.Now().UnixNano())
 		stats.MasterVolumeGrowthFrozen.Set(1)
 		glog.Errorf("volume growth frozen: a volume server reports topology id %s but this cluster is %s; the master state was rebuilt. Verify the cluster inventory, then run: volume.growth.resume", reported, own)
+		t.applyGrowthFreeze(true)
 	}
 }
 
@@ -1004,12 +1008,50 @@ func (t *Topology) IsGrowthFrozen() bool {
 }
 
 func (t *Topology) ResumeVolumeGrowth() bool {
+	if !t.growthFrozen.CompareAndSwap(true, false) {
+		return false
+	}
+	stats.MasterVolumeGrowthFrozen.Set(0)
+	glog.V(0).Infof("volume growth resumed by operator")
+	t.applyGrowthFreeze(false)
+	return true
+}
+
+// SetGrowthFrozen applies a replicated freeze state on every master. The
+// timestamp lands before the flag so a reader can never see frozen with a
+// zero time and misjudge the auto-resume window.
+func (t *Topology) SetGrowthFrozen(frozen bool) {
+	if frozen {
+		t.growthFrozenAt.Store(time.Now().UnixNano())
+		if t.growthFrozen.CompareAndSwap(false, true) {
+			stats.MasterVolumeGrowthFrozen.Set(1)
+			glog.V(0).Infof("volume growth freeze applied from raft")
+		}
+		return
+	}
 	if t.growthFrozen.CompareAndSwap(true, false) {
 		stats.MasterVolumeGrowthFrozen.Set(0)
-		glog.V(0).Infof("volume growth resumed by operator")
-		return true
 	}
-	return false
+}
+
+// applyGrowthFreeze replicates the freeze flag so followers keep it across a
+// leader change. No raft server (single-master test topologies) just sets it.
+func (t *Topology) applyGrowthFreeze(frozen bool) {
+	cmd := NewMaxVolumeIdCommand(t.GetMaxVolumeId(), t.GetTopologyId())
+	cmd.GrowthFrozen = &frozen
+	t.RaftServerAccessLock.RLock()
+	defer t.RaftServerAccessLock.RUnlock()
+	if t.RaftServer != nil {
+		if _, err := t.RaftServer.Do(cmd); err != nil {
+			glog.Warningf("apply growth freeze %v: %v", frozen, err)
+		}
+	} else if t.HashicorpRaft != nil {
+		if b, err := json.Marshal(cmd); err == nil {
+			t.HashicorpRaft.Apply(b, time.Second)
+		}
+	} else {
+		t.SetGrowthFrozen(frozen)
+	}
 }
 
 // DeferAllocation holds off new id allocation for d. Called when the leader
@@ -1043,6 +1085,9 @@ func (t *Topology) SetMaxVolumeIdFile(file string) {
 	if cmd.TopologyId != "" {
 		t.SetTopologyId(cmd.TopologyId)
 	}
+	if cmd.GrowthFrozen != nil {
+		t.SetGrowthFrozen(*cmd.GrowthFrozen)
+	}
 }
 
 // PersistMaxVolumeId writes the bound to the mirror file. Called by the
@@ -1066,7 +1111,12 @@ func (t *Topology) PersistMaxVolumeId(vid needle.VolumeId) error {
 		t.lastPersistedTopologyId.Store(topoId)
 		return nil
 	}
-	b, err := json.Marshal(NewMaxVolumeIdCommand(vid, topoId))
+	cmd := NewMaxVolumeIdCommand(vid, topoId)
+	if t.growthFrozen.Load() {
+		cmd.GrowthFrozen = new(bool)
+		*cmd.GrowthFrozen = true
+	}
+	b, err := json.Marshal(cmd)
 	if err != nil {
 		return err
 	}