From 6f7f5f91cb167b48f78645c7bdb2ce134ffaa6cf Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 18 Oct 2026 21:01:30 -0700 Subject: [PATCH] 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. --- weed/topology/max_volume_id_file_test.go | 21 +++++++++++++++++++++ weed/topology/topology.go | 14 +++++++++++--- 2 files changed, 32 insertions(+), 3 deletions(-) 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 }