topology: cover etcd allocator failures and mid-call bound growth (#11763)

weed/topology/volume_id_allocator_test.go

@@ -2,9 +2,11 @@ package topology
 
 import (
 	"context"
+	"errors"
 	"math"
 	"strconv"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -21,6 +23,11 @@ type fakeEtcdKV struct {
 	mu   sync.Mutex
 	rev  int64
 	data map[string]*mvccpb.KeyValue
+	// beforeCmp runs inside Commit before the compares are evaluated, to
+	// model a competing writer landing between the read and the CAS.
+	beforeCmp func()
+	getErr    error
+	commitErr error
 }
 
 func newFakeEtcdKV() *fakeEtcdKV {
@@ -30,6 +37,9 @@ func newFakeEtcdKV() *fakeEtcdKV {
 func (kv *fakeEtcdKV) Get(ctx context.Context, key string, opts ...clientv3.OpOption) (*clientv3.GetResponse, error) {
 	kv.mu.Lock()
 	defer kv.mu.Unlock()
+	if kv.getErr != nil {
+		return nil, kv.getErr
+	}
 	if v, ok := kv.data[key]; ok {
 		cp := *v
 		return &clientv3.GetResponse{Kvs: []*mvccpb.KeyValue{&cp}}, nil
@@ -100,6 +110,9 @@ func (t *fakeEtcdTxn) Else(ops ...clientv3.Op) clientv3.Txn {
 func (t *fakeEtcdTxn) Commit() (*clientv3.TxnResponse, error) {
 	t.kv.mu.Lock()
 	defer t.kv.mu.Unlock()
+	if t.kv.beforeCmp != nil {
+		t.kv.beforeCmp()
+	}
 	for _, cmp := range t.cmps {
 		if !t.kv.evalCmp(cmp) {
 			return &clientv3.TxnResponse{Succeeded: false}, nil
@@ -110,6 +123,9 @@ func (t *fakeEtcdTxn) Commit() (*clientv3.TxnResponse, error) {
 			t.kv.applyPut(string(op.KeyBytes()), string(op.ValueBytes()))
 		}
 	}
+	if t.kv.commitErr != nil {
+		return nil, t.kv.commitErr
+	}
 	return &clientv3.TxnResponse{Succeeded: true}, nil
 }
 
@@ -183,10 +199,12 @@ func TestEtcdAllocatorResumesFromStoredKey(t *testing.T) {
 
 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
+	// a competing writer lands the key once, between the read and the CAS
+	kv.beforeCmp = func() {
+		kv.applyPut("seaweedfs/max_volume_id", "55")
+		kv.beforeCmp = nil
+	}
 	alloc := newEtcdVolumeIdAllocator(kv, "seaweedfs", time.Second, func() needle.VolumeId {
-		kv.Put(context.Background(), "seaweedfs/max_volume_id", "55")
 		return 5
 	})
 	got, err := alloc.NextVolumeId()
@@ -198,6 +216,66 @@ func TestEtcdAllocatorRetriesOnCasConflict(t *testing.T) {
 	}
 }
 
+func TestEtcdAllocatorSkipsIdBelowFreshBound(t *testing.T) {
+	kv := newFakeEtcdKV()
+	var bound atomic.Uint32
+	bound.Store(5)
+	// a heartbeat raises the local bound while the CAS is in flight; the
+	// committed 6 must not be handed out
+	kv.beforeCmp = func() {
+		bound.Store(43)
+		kv.beforeCmp = nil
+	}
+	alloc := newEtcdVolumeIdAllocator(kv, "seaweedfs", time.Second, func() needle.VolumeId {
+		return needle.VolumeId(bound.Load())
+	})
+	got, err := alloc.NextVolumeId()
+	if err != nil {
+		t.Fatalf("alloc: %v", err)
+	}
+	if got != 44 {
+		t.Fatalf("an id at or below the fresher bound must be skipped, got %d", got)
+	}
+}
+
+func TestEtcdAllocatorReadError(t *testing.T) {
+	kv := newFakeEtcdKV()
+	kv.getErr = errors.New("etcd read failed")
+	alloc := newEtcdVolumeIdAllocator(kv, "seaweedfs", time.Second, func() needle.VolumeId {
+		return 41
+	})
+	if _, err := alloc.NextVolumeId(); err == nil {
+		t.Fatal("a failed read must return an error, not an id")
+	}
+	kv.getErr = nil
+	got, err := alloc.NextVolumeId()
+	if err != nil {
+		t.Fatalf("alloc: %v", err)
+	}
+	if got != 42 {
+		t.Fatalf("got %d, want 42", got)
+	}
+}
+
+func TestEtcdAllocatorCommitError(t *testing.T) {
+	kv := newFakeEtcdKV()
+	kv.commitErr = errors.New("etcd commit failed")
+	alloc := newEtcdVolumeIdAllocator(kv, "seaweedfs", time.Second, func() needle.VolumeId {
+		return 41
+	})
+	if _, err := alloc.NextVolumeId(); err == nil {
+		t.Fatal("a failed commit must return an error, not an id")
+	}
+	kv.commitErr = nil
+	got, err := alloc.NextVolumeId()
+	if err != nil {
+		t.Fatalf("alloc: %v", err)
+	}
+	if got != 43 {
+		t.Fatalf("the id written by the failed commit must be skipped, got %d", got)
+	}
+}
+
 func TestEtcdAllocatorExhaustion(t *testing.T) {
 	kv := newFakeEtcdKV()
 	kv.Put(context.Background(), "seaweedfs/max_volume_id", strconv.FormatUint(math.MaxUint32, 10))