From 0b3ed4502fc0c23bdc80fd40cbcd737f965b8098 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 19 Oct 2026 02:42:42 -0700 Subject: [PATCH] Design + master mirror of max volume id outside raft (#11763) (#11818) * design: durable volume-id allocation boundary (#11763) * master: mirror max volume id to a file outside raft (#11763) The allocation bound lived only in raft state, so a master that lost it rebuilt the bound from whatever volume servers were connected and could reissue ids held by offline servers. NextVolumeId now persists the bound to /max_volume_id before the raft apply, and both raft implementations persist on apply and recovery so follower mirrors stay current. The file also carries the topology id, which survives raft state cleanup the same way the snapshot recovery did. * master: serialize and fully sync the max volume id mirror (#11763) * master: rewrite the mirror when the topology id changes at the same bound A same-bound raft apply carrying a new topology id was skipped by the lastPersistedMaxVolumeId check, leaving the mirrored topology id empty until the next allocation. Track the mirrored topology id too and rewrite when either value differs. * drop the design doc from the PR stack --- weed/server/raft_hashicorp.go | 1 + weed/server/raft_server.go | 7 ++ weed/topology/cluster_commands.go | 3 + weed/topology/max_volume_id_file_test.go | 104 +++++++++++++++++++++++ weed/topology/topology.go | 85 ++++++++++++++++++ 5 files changed, 200 insertions(+) create mode 100644 weed/topology/max_volume_id_file_test.go diff --git a/weed/server/raft_hashicorp.go b/weed/server/raft_hashicorp.go index f64c0d26d7..9d1d601920 100644 --- a/weed/server/raft_hashicorp.go +++ b/weed/server/raft_hashicorp.go @@ -149,6 +149,7 @@ func NewHashicorpRaftServer(option *RaftServerOption) (*RaftServer, error) { if err := os.MkdirAll(path.Join(s.dataDir, "snapshots"), os.ModePerm); err != nil { return nil, err } + option.Topo.SetMaxVolumeIdFile(filepath.Join(s.dataDir, "max_volume_id")) baseDir := s.dataDir ldb, err := boltdb.NewBoltStore(filepath.Join(baseDir, ldbFile)) diff --git a/weed/server/raft_server.go b/weed/server/raft_server.go index 55d92254c1..8130b7838e 100644 --- a/weed/server/raft_server.go +++ b/weed/server/raft_server.go @@ -76,6 +76,9 @@ func (s StateMachine) Recovery(data []byte) error { s.topo.SetTopologyId(state.TopologyId) glog.V(0).Infof("Recovered TopologyId: %s", state.TopologyId) } + if err := s.topo.PersistMaxVolumeId(state.MaxVolumeId); err != nil { + glog.Warningf("persist max volume id %d: %v", state.MaxVolumeId, err) + } return nil } @@ -95,6 +98,9 @@ func (s *StateMachine) Apply(l *hashicorpRaft.Log) interface{} { glog.V(0).Infof("Set TopologyId from raft log: %s", state.TopologyId) } } + if err := s.topo.PersistMaxVolumeId(state.MaxVolumeId); err != nil { + glog.Warningf("persist max volume id %d: %v", state.MaxVolumeId, err) + } glog.V(1).Infoln("max volume id", before, "==>", s.topo.GetMaxVolumeId()) return nil @@ -163,6 +169,7 @@ func NewRaftServer(option *RaftServerOption) (*RaftServer, error) { if err := os.MkdirAll(path.Join(s.dataDir, "snapshot"), os.ModePerm); err != nil { return nil, err } + option.Topo.SetMaxVolumeIdFile(path.Join(s.dataDir, "max_volume_id")) stateMachine := StateMachine{topo: option.Topo} s.raftServer, err = raft.NewServer(string(s.serverAddr), s.dataDir, transporter, stateMachine, option.Topo, s.serverAddr.ToGrpcAddress()) diff --git a/weed/topology/cluster_commands.go b/weed/topology/cluster_commands.go index dfa3bf6cc5..1c713201e7 100644 --- a/weed/topology/cluster_commands.go +++ b/weed/topology/cluster_commands.go @@ -43,6 +43,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) + } glog.V(1).Infoln("max volume id", before, "==>", topo.GetMaxVolumeId()) return nil, nil diff --git a/weed/topology/max_volume_id_file_test.go b/weed/topology/max_volume_id_file_test.go new file mode 100644 index 0000000000..bdeff5c5aa --- /dev/null +++ b/weed/topology/max_volume_id_file_test.go @@ -0,0 +1,104 @@ +package topology + +import ( + "os" + "path/filepath" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/storage/needle" +) + +func TestMaxVolumeIdFileRestore(t *testing.T) { + dir := t.TempDir() + file := filepath.Join(dir, "max_volume_id") + + topo := NewTopology("topo", nil, 0, 5, false) + topo.SetMaxVolumeIdFile(file) + if err := topo.PersistMaxVolumeId(needle.VolumeId(41)); err != nil { + t.Fatalf("persist: %v", err) + } + + restarted := NewTopology("topo", nil, 0, 5, false) + restarted.SetMaxVolumeIdFile(file) + if got := restarted.GetMaxVolumeId(); got != needle.VolumeId(41) { + t.Fatalf("restored max volume id = %d, want 41", got) + } +} + +func TestMaxVolumeIdFileRestoresTopologyId(t *testing.T) { + dir := t.TempDir() + file := filepath.Join(dir, "max_volume_id") + + topo := NewTopology("topo", nil, 0, 5, false) + topo.SetTopologyId("cluster-a") + topo.SetMaxVolumeIdFile(file) + if err := topo.PersistMaxVolumeId(needle.VolumeId(7)); err != nil { + t.Fatalf("persist: %v", err) + } + + restarted := NewTopology("topo", nil, 0, 5, false) + restarted.SetMaxVolumeIdFile(file) + if got := restarted.GetTopologyId(); got != "cluster-a" { + t.Fatalf("restored topology id = %q, want cluster-a", got) + } +} + +func TestMaxVolumeIdFileRewritesOnTopologyIdChange(t *testing.T) { + dir := t.TempDir() + file := filepath.Join(dir, "max_volume_id") + + topo := NewTopology("topo", nil, 0, 5, false) + topo.SetMaxVolumeIdFile(file) + if err := topo.PersistMaxVolumeId(needle.VolumeId(7)); err != nil { + t.Fatalf("persist: %v", err) + } + topo.SetTopologyId("cluster-b") + if err := topo.PersistMaxVolumeId(needle.VolumeId(7)); err != nil { + t.Fatalf("persist same bound with new topology id: %v", err) + } + + restarted := NewTopology("topo", nil, 0, 5, false) + restarted.SetMaxVolumeIdFile(file) + if got := restarted.GetTopologyId(); got != "cluster-b" { + t.Fatalf("restored topology id = %q, want cluster-b", got) + } +} + +func TestMaxVolumeIdFileMonotonic(t *testing.T) { + dir := t.TempDir() + file := filepath.Join(dir, "max_volume_id") + + topo := NewTopology("topo", nil, 0, 5, false) + topo.SetMaxVolumeIdFile(file) + if err := topo.PersistMaxVolumeId(needle.VolumeId(10)); err != nil { + t.Fatalf("persist: %v", err) + } + if err := topo.PersistMaxVolumeId(needle.VolumeId(5)); err != nil { + t.Fatalf("persist lower: %v", err) + } + if got := topo.lastPersistedMaxVolumeId.Load(); got != 10 { + t.Fatalf("lastPersisted = %d, want 10", got) + } + + data, err := os.ReadFile(file) + if err != nil { + t.Fatalf("read file: %v", err) + } + if len(data) == 0 { + t.Fatal("mirror file is empty") + } +} + +func TestMaxVolumeIdFileCorrupt(t *testing.T) { + dir := t.TempDir() + file := filepath.Join(dir, "max_volume_id") + if err := os.WriteFile(file, []byte("{not json"), 0644); err != nil { + t.Fatal(err) + } + + topo := NewTopology("topo", nil, 0, 5, false) + topo.SetMaxVolumeIdFile(file) + if got := topo.GetMaxVolumeId(); got != 0 { + t.Fatalf("corrupt file seeded bound %d, want 0", got) + } +} diff --git a/weed/topology/topology.go b/weed/topology/topology.go index a892a0bad4..cd4d92ab37 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -6,6 +6,8 @@ import ( "fmt" "math" "math/rand/v2" + "os" + "path/filepath" "slices" "sync" "sync/atomic" @@ -81,6 +83,11 @@ type Topology struct { // the same node whether or not the grpc port suffix is present. dataNodeIndex map[string]*DataNode dataNodeIndexLock sync.RWMutex + + maxVolumeIdFile string + maxVolumeIdFileLock sync.Mutex + lastPersistedMaxVolumeId atomic.Uint32 + lastPersistedTopologyId atomic.Value } func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, pulse int, replicationAsMin bool) *Topology { @@ -390,6 +397,10 @@ func (t *Topology) NextVolumeId() (needle.VolumeId, error) { vid := t.GetMaxVolumeId() next := vid.Next() + if err := t.PersistMaxVolumeId(next); err != nil { + return 0, fmt.Errorf("persist max volume id %d: %w", next, err) + } + t.RaftServerAccessLock.RLock() defer t.RaftServerAccessLock.RUnlock() @@ -938,3 +949,77 @@ func (t *Topology) SetTopologyId(topologyId string) { glog.Fatalf("Split-brain detected! Current TopologyId is %s, but received %s. Stopping to prevent data corruption.", t.topologyId, topologyId) } } + +// 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. +func (t *Topology) SetMaxVolumeIdFile(file string) { + t.maxVolumeIdFile = file + data, err := os.ReadFile(file) + if err != nil { + return + } + cmd := &MaxVolumeIdCommand{} + if err := json.Unmarshal(data, cmd); err != nil { + glog.Warningf("corrupt max volume id file %s: %v", file, err) + return + } + t.lastPersistedMaxVolumeId.Store(uint32(cmd.MaxVolumeId)) + t.lastPersistedTopologyId.Store(cmd.TopologyId) + t.UpAdjustMaxVolumeId(cmd.MaxVolumeId) + if cmd.TopologyId != "" { + t.SetTopologyId(cmd.TopologyId) + } +} + +// PersistMaxVolumeId writes the bound to the mirror file. Called by the +// leader before a raft apply and by both raft implementations on apply and +// recovery, so follower mirrors stay current too. +func (t *Topology) PersistMaxVolumeId(vid needle.VolumeId) error { + topoId := t.GetTopologyId() + mirrored := func() bool { + return t.lastPersistedMaxVolumeId.Load() >= uint32(vid) && t.lastPersistedTopologyId.Load() == topoId + } + if mirrored() { + return nil + } + t.maxVolumeIdFileLock.Lock() + defer t.maxVolumeIdFileLock.Unlock() + if mirrored() { + return nil + } + if t.maxVolumeIdFile == "" { + t.lastPersistedMaxVolumeId.Store(uint32(vid)) + t.lastPersistedTopologyId.Store(topoId) + return nil + } + b, err := json.Marshal(NewMaxVolumeIdCommand(vid, topoId)) + if err != nil { + return err + } + tmp := t.maxVolumeIdFile + ".tmp" + f, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644) + if err != nil { + return err + } + if _, err := f.Write(b); err != nil { + f.Close() + return err + } + if err := f.Sync(); err != nil { + f.Close() + return err + } + if err := f.Close(); err != nil { + return err + } + if err := os.Rename(tmp, t.maxVolumeIdFile); err != nil { + return err + } + if err := util.FsyncDir(filepath.Dir(t.maxVolumeIdFile)); err != nil { + return err + } + t.lastPersistedMaxVolumeId.Store(uint32(vid)) + t.lastPersistedTopologyId.Store(topoId) + return nil +}