From f48c2d4ee1ffdb0289b59a39dc98f342326a794b 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 | 53 ++++++++++++----------- weed/topology/volume_id_allocator.go | 28 ++++++++++-- weed/topology/volume_id_allocator_test.go | 26 ++++++++--- 3 files changed, 74 insertions(+), 33 deletions(-) diff --git a/weed/topology/topology.go b/weed/topology/topology.go index 143539648c..fbb992e8a0 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -476,6 +476,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 } @@ -486,25 +491,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 @@ -1280,19 +1289,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 { - t.HashicorpRaft.Apply(b, time.Second) - } - } 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 {