From c2b7e50b2bf02f839123e75a5b6bd42cd446cff4 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 18 Oct 2026 20:59:44 -0700 Subject: [PATCH] allocator: never go below local bound, replicate external allocations via raft A stale etcd counter (restored backup or allocator re-enabled after being off) could hand out ids this master already knows are allocated; clamp the CAS base to max(stored, local seed) sampled once per call. Also apply the allocated id through raft so followers boundaries and mirrors track allocations made against the external allocator, and dedupe the raft apply path into raftApply. --- weed/topology/topology.go | 55 ++++++++++++----------- weed/topology/volume_id_allocator.go | 28 ++++++++++-- weed/topology/volume_id_allocator_test.go | 26 ++++++++--- 3 files changed, 74 insertions(+), 35 deletions(-) diff --git a/weed/topology/topology.go b/weed/topology/topology.go index f5e5e6acec..7821ae13f9 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -480,6 +480,11 @@ func (t *Topology) NextVolumeId() (needle.VolumeId, error) { if err := t.PersistMaxVolumeId(vid); err != nil { return 0, fmt.Errorf("can not assign next volume id %d: %w", vid, err) } + // Still record through raft so followers' boundaries and mirrors track + // allocations made against the external allocator. + if err := t.raftApply(NewMaxVolumeIdCommand(vid, t.GetTopologyId())); err != nil { + return 0, err + } t.UpAdjustMaxVolumeId(vid) return vid, nil } @@ -490,25 +495,29 @@ func (t *Topology) NextVolumeId() (needle.VolumeId, error) { return 0, fmt.Errorf("persist max volume id %d: %w", next, err) } - t.RaftServerAccessLock.RLock() - defer t.RaftServerAccessLock.RUnlock() - - if t.RaftServer != nil { - if _, err := t.RaftServer.Do(NewMaxVolumeIdCommand(next, t.GetTopologyId())); err != nil { - return 0, err - } - } else if t.HashicorpRaft != nil { - b, err := json.Marshal(NewMaxVolumeIdCommand(next, t.GetTopologyId())) - if err != nil { - return 0, fmt.Errorf("failed marshal NewMaxVolumeIdCommand: %+v", err) - } - if future := t.HashicorpRaft.Apply(b, time.Second); future.Error() != nil { - return 0, future.Error() - } + if err := t.raftApply(NewMaxVolumeIdCommand(next, t.GetTopologyId())); err != nil { + return 0, err } return next, nil } +func (t *Topology) raftApply(cmd *MaxVolumeIdCommand) error { + t.RaftServerAccessLock.RLock() + defer t.RaftServerAccessLock.RUnlock() + if t.RaftServer != nil { + _, err := t.RaftServer.Do(cmd) + return err + } + if t.HashicorpRaft != nil { + b, err := json.Marshal(cmd) + if err != nil { + return fmt.Errorf("failed marshal %s: %+v", cmd.CommandName(), err) + } + return t.HashicorpRaft.Apply(b, time.Second).Error() + } + return nil +} + // DefaultNeedleSizeEstimate is the fallback per-file-ID size estimate when // the client does not provide an expected data size. const DefaultNeedleSizeEstimate uint64 = 1024 * 1024 // 1 MB @@ -1297,21 +1306,13 @@ func (t *Topology) SetGrowthFrozen(frozen bool) { 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 { + if t.RaftServer != nil || t.HashicorpRaft != nil { + if err := t.raftApply(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) + return } + t.SetGrowthFrozen(frozen) } // SetTopologyInitPending gates allocation while a topology id is being diff --git a/weed/topology/volume_id_allocator.go b/weed/topology/volume_id_allocator.go index 695f236785..1f87452f53 100644 --- a/weed/topology/volume_id_allocator.go +++ b/weed/topology/volume_id_allocator.go @@ -25,12 +25,19 @@ type VolumeIdAllocator interface { // residual window: an id cannot be issued without the etcd write committing. type EtcdVolumeIdAllocator struct { kv clientv3.KV + client *clientv3.Client key string timeout time.Duration seed func() needle.VolumeId } -func NewEtcdVolumeIdAllocator(kv clientv3.KV, keyPrefix string, timeout time.Duration, seed func() needle.VolumeId) *EtcdVolumeIdAllocator { +func NewEtcdVolumeIdAllocator(client *clientv3.Client, keyPrefix string, timeout time.Duration, seed func() needle.VolumeId) *EtcdVolumeIdAllocator { + a := newEtcdVolumeIdAllocator(client.KV, keyPrefix, timeout, seed) + a.client = client + return a +} + +func newEtcdVolumeIdAllocator(kv clientv3.KV, keyPrefix string, timeout time.Duration, seed func() needle.VolumeId) *EtcdVolumeIdAllocator { return &EtcdVolumeIdAllocator{ kv: kv, key: strings.TrimRight(keyPrefix, "/") + "/max_volume_id", @@ -39,14 +46,28 @@ func NewEtcdVolumeIdAllocator(kv clientv3.KV, keyPrefix string, timeout time.Dur } } +func (a *EtcdVolumeIdAllocator) Close() error { + if a.client == nil { + return nil + } + return a.client.Close() +} + func (a *EtcdVolumeIdAllocator) NextVolumeId() (needle.VolumeId, error) { ctx, cancel := context.WithTimeout(context.Background(), a.timeout) defer cancel() + var seedBound uint64 + seeded := false for { resp, err := a.kv.Get(ctx, a.key) if err != nil { return 0, fmt.Errorf("volume id allocator read: %w", err) } + if !seeded { + // A stale etcd counter (restored backup, re-enabled allocator) must + // never go below the bound this master already knows. + seedBound, seeded = uint64(a.seed()), true + } var cur uint64 cmp := clientv3.Compare(clientv3.CreateRevision(a.key), "=", 0) if len(resp.Kvs) > 0 { @@ -54,8 +75,9 @@ func (a *EtcdVolumeIdAllocator) NextVolumeId() (needle.VolumeId, error) { return 0, fmt.Errorf("volume id counter %q: %w", resp.Kvs[0].Value, err) } cmp = clientv3.Compare(clientv3.ModRevision(a.key), "=", resp.Kvs[0].ModRevision) - } else { - cur = uint64(a.seed()) + } + if seedBound > cur { + cur = seedBound } if cur >= math.MaxUint32 { return 0, fmt.Errorf("volume id counter exhausted at %d", cur) diff --git a/weed/topology/volume_id_allocator_test.go b/weed/topology/volume_id_allocator_test.go index 39cbfbbb79..a3189a46b7 100644 --- a/weed/topology/volume_id_allocator_test.go +++ b/weed/topology/volume_id_allocator_test.go @@ -31,7 +31,8 @@ func (kv *fakeEtcdKV) Get(ctx context.Context, key string, opts ...clientv3.OpOp kv.mu.Lock() defer kv.mu.Unlock() if v, ok := kv.data[key]; ok { - return &clientv3.GetResponse{Kvs: []*mvccpb.KeyValue{v}}, nil + cp := *v + return &clientv3.GetResponse{Kvs: []*mvccpb.KeyValue{&cp}}, nil } return &clientv3.GetResponse{}, nil } @@ -136,7 +137,7 @@ func (kv *fakeEtcdKV) evalCmp(cmp clientv3.Cmp) bool { } func TestEtcdAllocatorSeedsFromLocalBound(t *testing.T) { - alloc := NewEtcdVolumeIdAllocator(newFakeEtcdKV(), "/sw/", time.Second, func() needle.VolumeId { + alloc := newEtcdVolumeIdAllocator(newFakeEtcdKV(), "/sw/", time.Second, func() needle.VolumeId { return 41 }) for i, want := range []needle.VolumeId{42, 43, 44} { @@ -150,10 +151,25 @@ func TestEtcdAllocatorSeedsFromLocalBound(t *testing.T) { } } +func TestEtcdAllocatorNeverBelowLocalBound(t *testing.T) { + kv := newFakeEtcdKV() + kv.Put(context.Background(), "seaweedfs/max_volume_id", "10") + alloc := newEtcdVolumeIdAllocator(kv, "seaweedfs", time.Second, func() needle.VolumeId { + return 41 + }) + got, err := alloc.NextVolumeId() + if err != nil { + t.Fatalf("alloc: %v", err) + } + if got != 42 { + t.Fatalf("a stale stored counter must not reissue ids: got %d", got) + } +} + func TestEtcdAllocatorResumesFromStoredKey(t *testing.T) { kv := newFakeEtcdKV() kv.Put(context.Background(), "seaweedfs/max_volume_id", "100") - alloc := NewEtcdVolumeIdAllocator(kv, "seaweedfs", time.Second, func() needle.VolumeId { + alloc := newEtcdVolumeIdAllocator(kv, "seaweedfs", time.Second, func() needle.VolumeId { return 41 }) got, err := alloc.NextVolumeId() @@ -169,7 +185,7 @@ func TestEtcdAllocatorRetriesOnCasConflict(t *testing.T) { kv := newFakeEtcdKV() // the seed runs after the empty read and a competing writer lands the key // before the allocator's create-CAS, forcing a retry - alloc := NewEtcdVolumeIdAllocator(kv, "seaweedfs", time.Second, func() needle.VolumeId { + alloc := newEtcdVolumeIdAllocator(kv, "seaweedfs", time.Second, func() needle.VolumeId { kv.Put(context.Background(), "seaweedfs/max_volume_id", "55") return 5 }) @@ -185,7 +201,7 @@ func TestEtcdAllocatorRetriesOnCasConflict(t *testing.T) { func TestEtcdAllocatorExhaustion(t *testing.T) { kv := newFakeEtcdKV() kv.Put(context.Background(), "seaweedfs/max_volume_id", strconv.FormatUint(math.MaxUint32, 10)) - alloc := NewEtcdVolumeIdAllocator(kv, "seaweedfs", time.Second, func() needle.VolumeId { + alloc := newEtcdVolumeIdAllocator(kv, "seaweedfs", time.Second, func() needle.VolumeId { return 5 }) if _, err := alloc.NextVolumeId(); err == nil {