diff --git a/weed/topology/max_volume_id_file_test.go b/weed/topology/max_volume_id_file_test.go index 4d2e7fef7c..bdeff5c5aa 100644 --- a/weed/topology/max_volume_id_file_test.go +++ b/weed/topology/max_volume_id_file_test.go @@ -43,6 +43,27 @@ func TestMaxVolumeIdFileRestoresTopologyId(t *testing.T) { } } +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") diff --git a/weed/topology/topology.go b/weed/topology/topology.go index 331135ed15..cd4d92ab37 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -87,6 +87,7 @@ type Topology struct { 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 { @@ -964,6 +965,7 @@ func (t *Topology) SetMaxVolumeIdFile(file string) { return } t.lastPersistedMaxVolumeId.Store(uint32(cmd.MaxVolumeId)) + t.lastPersistedTopologyId.Store(cmd.TopologyId) t.UpAdjustMaxVolumeId(cmd.MaxVolumeId) if cmd.TopologyId != "" { t.SetTopologyId(cmd.TopologyId) @@ -974,19 +976,24 @@ func (t *Topology) SetMaxVolumeIdFile(file string) { // 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 { - if t.lastPersistedMaxVolumeId.Load() >= uint32(vid) { + 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 t.lastPersistedMaxVolumeId.Load() >= uint32(vid) { + if mirrored() { return nil } if t.maxVolumeIdFile == "" { t.lastPersistedMaxVolumeId.Store(uint32(vid)) + t.lastPersistedTopologyId.Store(topoId) return nil } - b, err := json.Marshal(NewMaxVolumeIdCommand(vid, t.GetTopologyId())) + b, err := json.Marshal(NewMaxVolumeIdCommand(vid, topoId)) if err != nil { return err } @@ -1013,5 +1020,6 @@ func (t *Topology) PersistMaxVolumeId(vid needle.VolumeId) error { return err } t.lastPersistedMaxVolumeId.Store(uint32(vid)) + t.lastPersistedTopologyId.Store(topoId) return nil }