master: freeze volume growth on topology id mismatch (#11763)

weed/server/master_grpc_server.go

@@ -184,6 +184,7 @@ 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 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/master_grpc_server_volume.go

@@ -471,6 +471,13 @@ func (ms *MasterServer) VolumeMarkReadonly(ctx context.Context, req *master_pb.V
 	return resp, nil
 }
 
+func (ms *MasterServer) ResumeVolumeGrowth(ctx context.Context, req *master_pb.ResumeVolumeGrowthRequest) (*master_pb.ResumeVolumeGrowthResponse, error) {
+	if !ms.Topo.IsLeader() {
+		return nil, raft.NotLeaderError
+	}
+	return &master_pb.ResumeVolumeGrowthResponse{WasFrozen: ms.Topo.ResumeVolumeGrowth()}, nil
+}
+
 func (ms *MasterServer) VolumeGrow(ctx context.Context, req *master_pb.VolumeGrowRequest) (*master_pb.VolumeGrowResponse, error) {
 	if !ms.Topo.IsLeader() {
 		return nil, raft.NotLeaderError

weed/server/master_server.go

@@ -122,6 +122,8 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se
 	v.SetDefault("master.volume_growth.reservation_timeout", "5m")
 	v.SetDefault("master.volume_growth.disable", false)
 	option.VolumeGrowthDisabled = v.GetBool("master.volume_growth.disable")
+	v.SetDefault("master.volume_growth.freeze_on_topology_mismatch", true)
+	v.SetDefault("master.volume_growth.state_loss_recovery_wait", "0s")
 
 	topology.VolumeGrowStrategy.Copy1Count = v.GetUint32("master.volume_growth.copy_1")
 	topology.VolumeGrowStrategy.Copy2Count = v.GetUint32("master.volume_growth.copy_2")
@@ -157,6 +159,9 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se
 		glog.Fatalf("create sequencer failed.")
 	}
 	ms.Topo = topology.NewTopology("topo", seq, uint64(ms.option.VolumeSizeLimitMB)*1024*1024, 5, replicationAsMin)
+	ms.Topo.SetGrowthFreezeConfig(
+		v.GetBool("master.volume_growth.freeze_on_topology_mismatch"),
+		v.GetDuration("master.volume_growth.state_loss_recovery_wait"))
 	ms.vg = topology.NewDefaultVolumeGrowth()
 	glog.V(0).Infoln("Volume Size Limit is", ms.option.VolumeSizeLimitMB, "MB")
 

weed/server/raft_common.go

@@ -59,6 +59,7 @@ func EnsureTopologyId(topo *topology.Topology, checkLeaderFn func() bool, persis
 		} else {
 			glog.V(0).Infof("TopologyId generated: %s", topologyId)
 		}
+		topo.DeferAllocation(topo.WarmupDuration())
 		break
 	}
 }

weed/stats/metrics.go

@@ -101,6 +101,14 @@ var (
 			Help:      "admin lock",
 		}, []string{"client"})
 
+	MasterVolumeGrowthFrozen = prometheus.NewGauge(
+		prometheus.GaugeOpts{
+			Namespace: Namespace,
+			Subsystem: subsystemMaster,
+			Name:      "volume_growth_frozen",
+			Help:      "1 while volume growth is frozen after detected master state loss",
+		})
+
 	MasterReceivedHeartbeatCounter = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: Namespace,
@@ -991,6 +999,7 @@ func init() {
 	Gather.MustRegister(MasterClientConnectCounter)
 	Gather.MustRegister(MasterRaftIsleader)
 	Gather.MustRegister(MasterAdminLock)
+	Gather.MustRegister(MasterVolumeGrowthFrozen)
 	Gather.MustRegister(MasterReceivedHeartbeatCounter)
 	Gather.MustRegister(MasterLeaderChangeCounter)
 	Gather.MustRegister(MasterReplicaPlacementMismatch)

weed/topology/growth_freeze_test.go

@@ -0,0 +1,85 @@
+package topology
+
+import (
+	"testing"
+	"time"
+)
+
+func TestGrowthFreezeOnTopologyMismatch(t *testing.T) {
+	topo := NewTopology("topo", nil, 128, 3, false)
+	topo.SetGrowthFreezeConfig(true, 0)
+	topo.SetTopologyId("cluster-a")
+
+	topo.CheckReportedTopologyId("cluster-a")
+	if topo.IsGrowthFrozen() {
+		t.Fatal("matching topology id should not freeze growth")
+	}
+
+	topo.CheckReportedTopologyId("")
+	topo.CheckReportedTopologyId("cluster-b")
+	if !topo.IsGrowthFrozen() {
+		t.Fatal("mismatching topology id should freeze growth")
+	}
+
+	if !topo.ResumeVolumeGrowth() {
+		t.Fatal("resume should report the frozen state")
+	}
+	if topo.IsGrowthFrozen() {
+		t.Fatal("growth should be unfrozen after resume")
+	}
+	if topo.ResumeVolumeGrowth() {
+		t.Fatal("second resume should report no frozen state")
+	}
+}
+
+func TestGrowthFreezeDisabled(t *testing.T) {
+	topo := NewTopology("topo", nil, 128, 3, false)
+	topo.SetGrowthFreezeConfig(false, 0)
+	topo.SetTopologyId("cluster-a")
+
+	topo.CheckReportedTopologyId("cluster-b")
+	if topo.IsGrowthFrozen() {
+		t.Fatal("freeze_on_topology_mismatch=false should not freeze")
+	}
+}
+
+func TestGrowthFreezeAutoRecovery(t *testing.T) {
+	topo := NewTopology("topo", nil, 128, 3, false)
+	topo.SetGrowthFreezeConfig(true, 50*time.Millisecond)
+	topo.SetTopologyId("cluster-a")
+
+	topo.CheckReportedTopologyId("cluster-b")
+	if !topo.IsGrowthFrozen() {
+		t.Fatal("mismatching topology id should freeze growth")
+	}
+	time.Sleep(60 * time.Millisecond)
+	if topo.IsGrowthFrozen() {
+		t.Fatal("growth should auto-resume past the recovery wait")
+	}
+}
+
+func TestCheckReportedTopologyIdBeforeTopologyIdKnown(t *testing.T) {
+	topo := NewTopology("topo", nil, 128, 3, false)
+	topo.SetGrowthFreezeConfig(true, 0)
+
+	topo.CheckReportedTopologyId("cluster-b")
+	if topo.IsGrowthFrozen() {
+		t.Fatal("empty local topology id should not trigger the freeze")
+	}
+}
+
+func TestAllocationDelay(t *testing.T) {
+	topo := NewTopology("topo", nil, 128, 3, false)
+
+	if wait := topo.allocationDelay(); wait > 0 {
+		t.Fatalf("unexpected deferral: %v", wait)
+	}
+	topo.DeferAllocation(50 * time.Millisecond)
+	if wait := topo.allocationDelay(); wait <= 0 {
+		t.Fatal("expected a positive deferral after DeferAllocation")
+	}
+	topo.DeferAllocation(-time.Second)
+	if wait := topo.allocationDelay(); wait > 0 {
+		t.Fatalf("expired deferral should not delay: %v", wait)
+	}
+}

weed/topology/topology.go

@@ -88,6 +88,12 @@ type Topology struct {
 	maxVolumeIdFileLock      sync.Mutex
 	lastPersistedMaxVolumeId atomic.Uint32
 	lastPersistedTopologyId  atomic.Value
+
+	growthFrozen          atomic.Bool
+	growthFrozenAt        atomic.Int64
+	freezeOnMismatch      atomic.Bool
+	stateLossRecoveryWait atomic.Int64
+	allocationNotBefore   atomic.Int64
 }
 
 func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, pulse int, replicationAsMin bool) *Topology {
@@ -394,6 +400,13 @@ func (t *Topology) NextVolumeId() (needle.VolumeId, error) {
 		return 0, fmt.Errorf("as leader can not read yet")
 
 	}
+	if t.IsGrowthFrozen() {
+		return 0, fmt.Errorf("volume growth is frozen after master state loss; verify the cluster inventory, then run: volume.growth.resume")
+	}
+	if wait := t.allocationDelay(); wait > 0 {
+		glog.V(0).Infof("deferring volume id allocation for %v", wait)
+		time.Sleep(wait)
+	}
 	vid := t.GetMaxVolumeId()
 	next := vid.Next()
 
@@ -950,6 +963,66 @@ func (t *Topology) SetTopologyId(topologyId string) {
 	}
 }
 
+func (t *Topology) SetGrowthFreezeConfig(freezeOnMismatch bool, recoveryWait time.Duration) {
+	t.freezeOnMismatch.Store(freezeOnMismatch)
+	t.stateLossRecoveryWait.Store(int64(recoveryWait))
+}
+
+// CheckReportedTopologyId freezes growth when a volume server reports a
+// topology id different from this cluster's: the topology id survives raft
+// state cleanup, so a mismatch means the master's state was rebuilt (or a
+// foreign disk was cloned in).
+func (t *Topology) CheckReportedTopologyId(reported string) {
+	if reported == "" || !t.freezeOnMismatch.Load() {
+		return
+	}
+	own := t.GetTopologyId()
+	if own == "" || reported == own {
+		return
+	}
+	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)
+	}
+}
+
+func (t *Topology) IsGrowthFrozen() bool {
+	if !t.growthFrozen.Load() {
+		return false
+	}
+	if wait := t.stateLossRecoveryWait.Load(); wait > 0 {
+		if time.Duration(time.Now().UnixNano()-t.growthFrozenAt.Load()) > time.Duration(wait) {
+			if t.growthFrozen.CompareAndSwap(true, false) {
+				stats.MasterVolumeGrowthFrozen.Set(0)
+				glog.V(0).Infof("volume growth unfrozen after %v", time.Duration(wait))
+			}
+			return false
+		}
+	}
+	return true
+}
+
+func (t *Topology) ResumeVolumeGrowth() bool {
+	if t.growthFrozen.CompareAndSwap(true, false) {
+		stats.MasterVolumeGrowthFrozen.Set(0)
+		glog.V(0).Infof("volume growth resumed by operator")
+		return true
+	}
+	return false
+}
+
+// DeferAllocation holds off new id allocation for d. Called when the leader
+// 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())
+}
+
+func (t *Topology) allocationDelay() time.Duration {
+	return time.Until(time.Unix(0, t.allocationNotBefore.Load()))
+}
+
 // SetMaxVolumeIdFile mirrors the allocation bound to a file outside raft, so
 // a master whose raft state is gone still restores a safe boundary. An
 // existing file seeds both the bound and the topology id.