Freeze volume growth on topology id mismatch (#11763) (#11820)

* pb: ResumeVolumeGrowth rpc for operator-driven growth resume (#11763)

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

* shell: volume.growth.resume command (#11763)

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

* master: block allocation while the topology id is being minted

DeferAllocation only armed after the topology id persisted, so growth
requests arriving during init ran ahead of the warmup window. Gate
NextVolumeId on an init-pending flag spanning the whole mint.

* master: stash reported topology ids until the local id is minted (#11763)

* drain stashed topology id reports when the id arrives outside the mint path

* persist the growth-freeze flag in the mirror file and raft snapshots

* log hashicorp raft marshal and apply errors in applyGrowthFreeze

seaweed-volume/proto/master.proto

@@ -59,6 +59,8 @@ service Seaweed {
   }
   rpc RaftLeadershipTransfer (RaftLeadershipTransferRequest) returns (RaftLeadershipTransferResponse) {
   }
+  rpc ResumeVolumeGrowth (ResumeVolumeGrowthRequest) returns (ResumeVolumeGrowthResponse) {
+  }
   rpc VolumeGrow (VolumeGrowRequest) returns (VolumeGrowResponse) {
   }
   rpc CollectionStatistics (CollectionStatisticsRequest) returns (CollectionStatisticsResponse) {
@@ -324,6 +326,13 @@ message VolumeGrowRequest {
   float grow_avg_fill_threshold = 10;
 }
 
+message ResumeVolumeGrowthRequest {
+}
+
+message ResumeVolumeGrowthResponse {
+  bool was_frozen = 1;
+}
+
 message AssignResponse {
   string fid = 1;
   uint64 count = 4;

weed/pb/master.proto

@@ -59,6 +59,8 @@ service Seaweed {
   }
   rpc RaftLeadershipTransfer (RaftLeadershipTransferRequest) returns (RaftLeadershipTransferResponse) {
   }
+  rpc ResumeVolumeGrowth (ResumeVolumeGrowthRequest) returns (ResumeVolumeGrowthResponse) {
+  }
   rpc VolumeGrow (VolumeGrowRequest) returns (VolumeGrowResponse) {
   }
   rpc CollectionStatistics (CollectionStatisticsRequest) returns (CollectionStatisticsResponse) {
@@ -324,6 +326,13 @@ message VolumeGrowRequest {
   float grow_avg_fill_threshold = 10;
 }
 
+message ResumeVolumeGrowthRequest {
+}
+
+message ResumeVolumeGrowthResponse {
+  bool was_frozen = 1;
+}
+
 message AssignResponse {
   string fid = 1;
   uint64 count = 4;

weed/pb/master_pb/master_grpc.pb.go

@@ -44,6 +44,7 @@ const (
 	Seaweed_RaftAddServer_FullMethodName          = "/master_pb.Seaweed/RaftAddServer"
 	Seaweed_RaftRemoveServer_FullMethodName       = "/master_pb.Seaweed/RaftRemoveServer"
 	Seaweed_RaftLeadershipTransfer_FullMethodName = "/master_pb.Seaweed/RaftLeadershipTransfer"
+	Seaweed_ResumeVolumeGrowth_FullMethodName     = "/master_pb.Seaweed/ResumeVolumeGrowth"
 	Seaweed_VolumeGrow_FullMethodName             = "/master_pb.Seaweed/VolumeGrow"
 	Seaweed_CollectionStatistics_FullMethodName   = "/master_pb.Seaweed/CollectionStatistics"
 )
@@ -77,6 +78,7 @@ type SeaweedClient interface {
 	RaftAddServer(ctx context.Context, in *RaftAddServerRequest, opts ...grpc.CallOption) (*RaftAddServerResponse, error)
 	RaftRemoveServer(ctx context.Context, in *RaftRemoveServerRequest, opts ...grpc.CallOption) (*RaftRemoveServerResponse, error)
 	RaftLeadershipTransfer(ctx context.Context, in *RaftLeadershipTransferRequest, opts ...grpc.CallOption) (*RaftLeadershipTransferResponse, error)
+	ResumeVolumeGrowth(ctx context.Context, in *ResumeVolumeGrowthRequest, opts ...grpc.CallOption) (*ResumeVolumeGrowthResponse, error)
 	VolumeGrow(ctx context.Context, in *VolumeGrowRequest, opts ...grpc.CallOption) (*VolumeGrowResponse, error)
 	CollectionStatistics(ctx context.Context, in *CollectionStatisticsRequest, opts ...grpc.CallOption) (*CollectionStatisticsResponse, error)
 }
@@ -357,6 +359,16 @@ func (c *seaweedClient) RaftLeadershipTransfer(ctx context.Context, in *RaftLead
 	return out, nil
 }
 
+func (c *seaweedClient) ResumeVolumeGrowth(ctx context.Context, in *ResumeVolumeGrowthRequest, opts ...grpc.CallOption) (*ResumeVolumeGrowthResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(ResumeVolumeGrowthResponse)
+	err := c.cc.Invoke(ctx, Seaweed_ResumeVolumeGrowth_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 func (c *seaweedClient) VolumeGrow(ctx context.Context, in *VolumeGrowRequest, opts ...grpc.CallOption) (*VolumeGrowResponse, error) {
 	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
 	out := new(VolumeGrowResponse)
@@ -406,6 +418,7 @@ type SeaweedServer interface {
 	RaftAddServer(context.Context, *RaftAddServerRequest) (*RaftAddServerResponse, error)
 	RaftRemoveServer(context.Context, *RaftRemoveServerRequest) (*RaftRemoveServerResponse, error)
 	RaftLeadershipTransfer(context.Context, *RaftLeadershipTransferRequest) (*RaftLeadershipTransferResponse, error)
+	ResumeVolumeGrowth(context.Context, *ResumeVolumeGrowthRequest) (*ResumeVolumeGrowthResponse, error)
 	VolumeGrow(context.Context, *VolumeGrowRequest) (*VolumeGrowResponse, error)
 	CollectionStatistics(context.Context, *CollectionStatisticsRequest) (*CollectionStatisticsResponse, error)
 	mustEmbedUnimplementedSeaweedServer()
@@ -493,6 +506,9 @@ func (UnimplementedSeaweedServer) RaftRemoveServer(context.Context, *RaftRemoveS
 func (UnimplementedSeaweedServer) RaftLeadershipTransfer(context.Context, *RaftLeadershipTransferRequest) (*RaftLeadershipTransferResponse, error) {
 	return nil, status.Error(codes.Unimplemented, "method RaftLeadershipTransfer not implemented")
 }
+func (UnimplementedSeaweedServer) ResumeVolumeGrowth(context.Context, *ResumeVolumeGrowthRequest) (*ResumeVolumeGrowthResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method ResumeVolumeGrowth not implemented")
+}
 func (UnimplementedSeaweedServer) VolumeGrow(context.Context, *VolumeGrowRequest) (*VolumeGrowResponse, error) {
 	return nil, status.Error(codes.Unimplemented, "method VolumeGrow not implemented")
 }
@@ -930,6 +946,24 @@ func _Seaweed_RaftLeadershipTransfer_Handler(srv interface{}, ctx context.Contex
 	return interceptor(ctx, in, info, handler)
 }
 
+func _Seaweed_ResumeVolumeGrowth_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ResumeVolumeGrowthRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(SeaweedServer).ResumeVolumeGrowth(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: Seaweed_ResumeVolumeGrowth_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(SeaweedServer).ResumeVolumeGrowth(ctx, req.(*ResumeVolumeGrowthRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 func _Seaweed_VolumeGrow_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(VolumeGrowRequest)
 	if err := dec(in); err != nil {
@@ -1057,6 +1091,10 @@ var Seaweed_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "RaftLeadershipTransfer",
 			Handler:    _Seaweed_RaftLeadershipTransfer_Handler,
 		},
+		{
+			MethodName: "ResumeVolumeGrowth",
+			Handler:    _Seaweed_ResumeVolumeGrowth_Handler,
+		},
 		{
 			MethodName: "VolumeGrow",
 			Handler:    _Seaweed_VolumeGrow_Handler,

weed/server/master_grpc_server.go

@@ -184,6 +184,9 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 
 		ms.Topo.Sequence.SetMax(heartbeat.MaxFileKey)
 		ms.Topo.UpAdjustMaxVolumeId(needle.VolumeId(heartbeat.MaxVolumeIdSeen))
+		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/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

@@ -30,9 +30,15 @@ func recoverTopologyIdFromState(fsmState []byte, topo *topology.Topology) {
 // It uses the provided checkLeaderFn to verify leadership and persistFn to save the new ID.
 func EnsureTopologyId(topo *topology.Topology, checkLeaderFn func() bool, persistFn func(string) error) {
 	if topo.GetTopologyId() != "" {
+		// The id arrived through raft replay or the mirror file; reports
+		// stashed while it was unknown can now be checked.
+		topo.SetTopologyInitPending(false)
 		return
 	}
 
+	topo.SetTopologyInitPending(true)
+	defer topo.SetTopologyInitPending(false)
+
 	topologyId := uuid.New().String()
 	for {
 		if !checkLeaderFn() {
@@ -59,6 +65,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/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)
 }
@@ -76,6 +80,9 @@ func (s StateMachine) Recovery(data []byte) error {
 		s.topo.SetTopologyId(state.TopologyId)
 		glog.V(0).Infof("Recovered TopologyId: %s", state.TopologyId)
 	}
+	if state.GrowthFrozen != nil {
+		s.topo.SetGrowthFrozen(*state.GrowthFrozen)
+	}
 	if err := s.topo.PersistMaxVolumeId(state.MaxVolumeId); err != nil {
 		glog.Warningf("persist max volume id %d: %v", state.MaxVolumeId, err)
 	}
@@ -98,6 +105,9 @@ func (s *StateMachine) Apply(l *hashicorpRaft.Log) interface{} {
 			glog.V(0).Infof("Set TopologyId from raft log: %s", state.TopologyId)
 		}
 	}
+	if state.GrowthFrozen != nil {
+		s.topo.SetGrowthFrozen(*state.GrowthFrozen)
+	}
 	if err := s.topo.PersistMaxVolumeId(state.MaxVolumeId); err != nil {
 		glog.Warningf("persist max volume id %d: %v", state.MaxVolumeId, err)
 	}
@@ -107,10 +117,13 @@ func (s *StateMachine) Apply(l *hashicorpRaft.Log) interface{} {
 }
 
 func (s *StateMachine) Snapshot() (hashicorpRaft.FSMSnapshot, error) {
-	return &topology.MaxVolumeIdCommand{
-		MaxVolumeId: s.topo.GetMaxVolumeId(),
-		TopologyId:  s.topo.GetTopologyId(),
-	}, nil
+	frozen := s.topo.IsGrowthFrozen()
+	state := &topology.MaxVolumeIdCommand{
+		MaxVolumeId:  s.topo.GetMaxVolumeId(),
+		TopologyId:   s.topo.GetTopologyId(),
+		GrowthFrozen: &frozen,
+	}
+	return state, nil
 }
 
 func (s *StateMachine) Restore(r io.ReadCloser) error {

weed/shell/command_volume_growth_resume.go

@@ -0,0 +1,52 @@
+package shell
+
+import (
+	"context"
+	"fmt"
+	"io"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+)
+
+func init() {
+	Commands = append(Commands, &commandVolumeGrowthResume{})
+}
+
+type commandVolumeGrowthResume struct {
+}
+
+func (c *commandVolumeGrowthResume) Name() string {
+	return "volume.growth.resume"
+}
+
+func (c *commandVolumeGrowthResume) Help() string {
+	return `resume volume growth frozen after detected master state loss
+
+	volume.growth.resume
+
+	Volume growth freezes automatically when a volume server reports a
+	topology id that differs from this cluster's, which means the master
+	state was rebuilt and max volume id knowledge may be incomplete. Run
+	this after verifying the cluster inventory so new ids are not reused.
+
+`
+}
+
+func (c *commandVolumeGrowthResume) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandVolumeGrowthResume) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
+	return commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
+		resp, err := client.ResumeVolumeGrowth(context.Background(), &master_pb.ResumeVolumeGrowthRequest{})
+		if err != nil {
+			return err
+		}
+		if resp.WasFrozen {
+			fmt.Fprintf(writer, "volume growth resumed\n")
+		} else {
+			fmt.Fprintf(writer, "volume growth was not frozen\n")
+		}
+		return nil
+	})
+}

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/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 {
@@ -43,6 +46,9 @@ func (c *MaxVolumeIdCommand) Apply(server raft.Server) (interface{}, error) {
 			}
 		}
 	}
+	if c.GrowthFrozen != nil {
+		topo.SetGrowthFrozen(*c.GrowthFrozen)
+	}
 	if err := topo.PersistMaxVolumeId(c.MaxVolumeId); err != nil {
 		glog.Warningf("persist max volume id %d: %v", c.MaxVolumeId, err)
 	}

weed/topology/growth_freeze_test.go

@@ -0,0 +1,146 @@
+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")
+	}
+
+	topo.SetTopologyId("cluster-c")
+	topo.SetTopologyInitPending(false)
+	if !topo.IsGrowthFrozen() {
+		t.Fatal("a report stashed before the topology id was known should still freeze growth")
+	}
+}
+
+func TestReportedTopologyIdStashDoesNotFreezeOnMatch(t *testing.T) {
+	topo := NewTopology("topo", nil, 128, 3, false)
+	topo.SetGrowthFreezeConfig(true, 0)
+
+	topo.CheckReportedTopologyId("cluster-a")
+	topo.SetTopologyId("cluster-a")
+	topo.SetTopologyInitPending(false)
+	if topo.IsGrowthFrozen() {
+		t.Fatal("a stashed report matching the minted id should not freeze growth")
+	}
+}
+
+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)
+
+	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

@@ -84,10 +84,21 @@ type Topology struct {
 	dataNodeIndex     map[string]*DataNode
 	dataNodeIndexLock sync.RWMutex
 
-	maxVolumeIdFile          string
-	maxVolumeIdFileLock      sync.Mutex
-	lastPersistedMaxVolumeId atomic.Uint32
-	lastPersistedTopologyId  atomic.Value
+	maxVolumeIdFile           string
+	maxVolumeIdFileLock       sync.Mutex
+	lastPersistedMaxVolumeId  atomic.Uint32
+	lastPersistedTopologyId   atomic.Value
+	lastPersistedGrowthFrozen atomic.Bool
+
+	growthFrozen          atomic.Bool
+	growthFrozenAt        atomic.Int64
+	freezeOnMismatch      atomic.Bool
+	stateLossRecoveryWait atomic.Int64
+	allocationNotBefore   atomic.Int64
+	topologyInitPending   atomic.Bool
+
+	reportedTopologyIdsLock    sync.Mutex
+	pendingReportedTopologyIds map[string]struct{}
 }
 
 func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, pulse int, replicationAsMin bool) *Topology {
@@ -394,6 +405,22 @@ func (t *Topology) NextVolumeId() (needle.VolumeId, error) {
 		return 0, fmt.Errorf("as leader can not read yet")
 
 	}
+	for t.topologyInitPending.Load() {
+		if !t.IsLeaderAndCanRead() {
+			return 0, fmt.Errorf("as leader can not read yet")
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+	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)
+		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()
 
@@ -950,6 +977,135 @@ 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 == "" {
+		// The local topology id is still being minted. Keep the report so a
+		// later heartbeat echo cannot erase the mismatch on the server side
+		// before the freeze check ever ran.
+		t.reportedTopologyIdsLock.Lock()
+		if t.pendingReportedTopologyIds == nil {
+			t.pendingReportedTopologyIds = make(map[string]struct{})
+		}
+		t.pendingReportedTopologyIds[reported] = struct{}{}
+		t.reportedTopologyIdsLock.Unlock()
+		return
+	}
+	if reported == own {
+		return
+	}
+	t.growthFrozenAt.Store(time.Now().UnixNano())
+	if t.growthFrozen.CompareAndSwap(false, true) {
+		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)
+	}
+}
+
+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) {
+		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)
+	}
+}
+
+// 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 {
+			glog.Warningf("marshal growth freeze %v: %v", frozen, err)
+		} else if future := t.HashicorpRaft.Apply(b, time.Second); future.Error() != nil {
+			glog.Warningf("apply growth freeze %v: %v", frozen, future.Error())
+		}
+	} else {
+		t.SetGrowthFrozen(frozen)
+	}
+}
+
+// SetTopologyInitPending gates allocation while a topology id is being
+// minted, so growth cannot slip in before the warmup window is installed.
+func (t *Topology) SetTopologyInitPending(pending bool) {
+	t.topologyInitPending.Store(pending)
+	if pending {
+		return
+	}
+	t.reportedTopologyIdsLock.Lock()
+	reported := t.pendingReportedTopologyIds
+	t.pendingReportedTopologyIds = nil
+	t.reportedTopologyIdsLock.Unlock()
+	for id := range reported {
+		t.CheckReportedTopologyId(id)
+	}
+}
+
+// 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.
@@ -970,6 +1126,10 @@ func (t *Topology) SetMaxVolumeIdFile(file string) {
 	if cmd.TopologyId != "" {
 		t.SetTopologyId(cmd.TopologyId)
 	}
+	if cmd.GrowthFrozen != nil {
+		t.lastPersistedGrowthFrozen.Store(*cmd.GrowthFrozen)
+		t.SetGrowthFrozen(*cmd.GrowthFrozen)
+	}
 }
 
 // PersistMaxVolumeId writes the bound to the mirror file. Called by the
@@ -977,8 +1137,9 @@ func (t *Topology) SetMaxVolumeIdFile(file string) {
 // recovery, so follower mirrors stay current too.
 func (t *Topology) PersistMaxVolumeId(vid needle.VolumeId) error {
 	topoId := t.GetTopologyId()
+	frozen := t.IsGrowthFrozen()
 	mirrored := func() bool {
-		return t.lastPersistedMaxVolumeId.Load() >= uint32(vid) && t.lastPersistedTopologyId.Load() == topoId
+		return t.lastPersistedMaxVolumeId.Load() >= uint32(vid) && t.lastPersistedTopologyId.Load() == topoId && t.lastPersistedGrowthFrozen.Load() == frozen
 	}
 	if mirrored() {
 		return nil
@@ -991,9 +1152,12 @@ func (t *Topology) PersistMaxVolumeId(vid needle.VolumeId) error {
 	if t.maxVolumeIdFile == "" {
 		t.lastPersistedMaxVolumeId.Store(uint32(vid))
 		t.lastPersistedTopologyId.Store(topoId)
+		t.lastPersistedGrowthFrozen.Store(frozen)
 		return nil
 	}
-	b, err := json.Marshal(NewMaxVolumeIdCommand(vid, topoId))
+	cmd := NewMaxVolumeIdCommand(vid, topoId)
+	cmd.GrowthFrozen = &frozen
+	b, err := json.Marshal(cmd)
 	if err != nil {
 		return err
 	}
@@ -1021,5 +1185,6 @@ func (t *Topology) PersistMaxVolumeId(vid needle.VolumeId) error {
 	}
 	t.lastPersistedMaxVolumeId.Store(uint32(vid))
 	t.lastPersistedTopologyId.Store(topoId)
+	t.lastPersistedGrowthFrozen.Store(frozen)
 	return nil
 }