mirror of
https://github.com/tiennm99/miti99bot.git
synced 2026-09-08 20:20:17 +00:00
fix(storage): report missing CAS support as unsupported instead of conflict
- prefix.go: return errors.ErrUnsupported when inner store lacks CompareAndSwap, enabling fail-fast instead of infinite retries - Adds comprehensive CAS semantics test coverage for all KV backends: * memory_kv_test.go: new tests for basic operations and CAS failures * prefix_test.go: tests for wrapped CAS errors and unsupported operations * dynamodb_kv_test.go, firestore_kv_test.go: CAS failure scenarios - portfolio_test.go: test retry exhaustion, business-error short-circuit, concurrent updates, and fail-fast on unsupported CAS
This commit is contained in:
@@ -2,7 +2,10 @@ package gold
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"math"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
|
||||
stocktrading "github.com/tiennm99/miti99bot/internal/modules/trading"
|
||||
@@ -108,6 +111,136 @@ func TestUpdatePortfolioRetriesAfterWriteConflict(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
type alwaysConflictStore struct {
|
||||
storage.KVStore
|
||||
attempts int
|
||||
}
|
||||
|
||||
func (s *alwaysConflictStore) CompareAndSwap(context.Context, string, []byte, []byte) error {
|
||||
s.attempts++
|
||||
return storage.ErrConflict
|
||||
}
|
||||
|
||||
func TestUpdatePortfolioReturnsConflictAfterExhaustingRetries(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
kv := &alwaysConflictStore{KVStore: storage.NewMemoryKVStore()}
|
||||
_, err := UpdatePortfolio(ctx, kv, 7, 1, func(p *Portfolio) error {
|
||||
p.AddVND(5)
|
||||
return nil
|
||||
})
|
||||
if !errors.Is(err, storage.ErrConflict) {
|
||||
t.Fatalf("UpdatePortfolio: got %v, want wrapped ErrConflict", err)
|
||||
}
|
||||
if kv.attempts != portfolioUpdateAttempts {
|
||||
t.Errorf("CAS attempts = %d, want %d", kv.attempts, portfolioUpdateAttempts)
|
||||
}
|
||||
if _, err := kv.KVStore.Get(ctx, "user:7"); !errors.Is(err, storage.ErrNotFound) {
|
||||
t.Errorf("exhausted update must not persist anything, Get = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
type countingCASStore struct {
|
||||
*storage.MemoryKVStore
|
||||
casCalls int
|
||||
}
|
||||
|
||||
func (s *countingCASStore) CompareAndSwap(ctx context.Context, key string, expected, val []byte) error {
|
||||
s.casCalls++
|
||||
return s.MemoryKVStore.CompareAndSwap(ctx, key, expected, val)
|
||||
}
|
||||
|
||||
func TestUpdatePortfolioMutateErrorDoesNotRetryOrPersist(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
kv := &countingCASStore{MemoryKVStore: storage.NewMemoryKVStore()}
|
||||
mutateCalls := 0
|
||||
_, err := UpdatePortfolio(ctx, kv, 7, 1, func(p *Portfolio) error {
|
||||
mutateCalls++
|
||||
return errInsufficientVND
|
||||
})
|
||||
if !errors.Is(err, errInsufficientVND) {
|
||||
t.Fatalf("UpdatePortfolio: got %v, want errInsufficientVND", err)
|
||||
}
|
||||
if mutateCalls != 1 {
|
||||
t.Errorf("mutate calls = %d, want 1 (business errors must not retry)", mutateCalls)
|
||||
}
|
||||
if kv.casCalls != 0 {
|
||||
t.Errorf("CAS calls = %d, want 0 (failed mutate must not write)", kv.casCalls)
|
||||
}
|
||||
if _, err := kv.Get(ctx, "user:7"); !errors.Is(err, storage.ErrNotFound) {
|
||||
t.Errorf("failed mutate must not persist anything, Get = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUpdatePortfolioConcurrentIncrementsLoseNoUpdates(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
kv := storage.NewMemoryKVStore()
|
||||
const goroutines = 16
|
||||
var successes atomic.Int64
|
||||
var wg sync.WaitGroup
|
||||
errs := make(chan error, goroutines)
|
||||
for range goroutines {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
_, err := UpdatePortfolio(ctx, kv, 7, 1, func(p *Portfolio) error {
|
||||
p.AddVND(1000)
|
||||
return nil
|
||||
})
|
||||
if err == nil {
|
||||
successes.Add(1)
|
||||
} else if !errors.Is(err, storage.ErrConflict) {
|
||||
errs <- err
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
close(errs)
|
||||
for err := range errs {
|
||||
t.Fatalf("UpdatePortfolio: %v", err)
|
||||
}
|
||||
// Retry exhaustion under heavy contention is acceptable; a lost update is not:
|
||||
// the stored balance must equal exactly 1000 per successful update.
|
||||
if successes.Load() == 0 {
|
||||
t.Fatal("no update succeeded")
|
||||
}
|
||||
loaded, err := LoadPortfolio(ctx, kv, 7, 1)
|
||||
if err != nil {
|
||||
t.Fatalf("LoadPortfolio: %v", err)
|
||||
}
|
||||
if want := float64(successes.Load()) * 1000; loaded.VND != want {
|
||||
t.Errorf("VND = %v, want %v (%d successful updates)", loaded.VND, want, successes.Load())
|
||||
}
|
||||
}
|
||||
|
||||
// plainKVStore hides the embedded store's CompareAndSwap method to model a
|
||||
// backend without CAS support.
|
||||
type plainKVStore struct {
|
||||
storage.KVStore
|
||||
}
|
||||
|
||||
func TestUpdatePortfolioFailsFastWhenCASUnsupported(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
mutate := func(p *Portfolio) error {
|
||||
p.AddVND(5)
|
||||
return nil
|
||||
}
|
||||
|
||||
// A bare non-CAS store is rejected by the capability check.
|
||||
if _, err := UpdatePortfolio(ctx, &plainKVStore{KVStore: storage.NewMemoryKVStore()}, 7, 1, mutate); err == nil {
|
||||
t.Error("UpdatePortfolio on non-CAS store: want error, got nil")
|
||||
}
|
||||
|
||||
// A prefixed wrapper around a non-CAS store must fail fast as unsupported,
|
||||
// not burn retries on a fake conflict.
|
||||
_, err := UpdatePortfolio(ctx, storage.Prefixed(&plainKVStore{KVStore: storage.NewMemoryKVStore()}, "gold"), 7, 1, mutate)
|
||||
if !errors.Is(err, errors.ErrUnsupported) {
|
||||
t.Errorf("got %v, want errors.ErrUnsupported", err)
|
||||
}
|
||||
if errors.Is(err, storage.ErrConflict) {
|
||||
t.Error("missing CAS capability must not be reported as a write conflict")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTradingAndGoldPortfolioKeysDoNotCollide(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
provider := storage.NewMemoryProvider()
|
||||
|
||||
@@ -152,6 +152,42 @@ func TestDynamoDBKVStore_ListPrefix(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestDynamoDBKVStore_CompareAndSwap(t *testing.T) {
|
||||
client, table, cleanup := dynamoDBLocalSetup(t)
|
||||
defer cleanup()
|
||||
|
||||
ctx := context.Background()
|
||||
s := NewDynamoDBKVStore(client, table, "gold")
|
||||
|
||||
// Create-if-absent succeeds once, then conflicts.
|
||||
if err := s.CompareAndSwap(ctx, "user:1", nil, []byte("v1")); err != nil {
|
||||
t.Fatalf("CompareAndSwap create: %v", err)
|
||||
}
|
||||
if err := s.CompareAndSwap(ctx, "user:1", nil, []byte("v2")); !errors.Is(err, ErrConflict) {
|
||||
t.Errorf("CompareAndSwap create over existing: got %v, want ErrConflict", err)
|
||||
}
|
||||
|
||||
// Swap with matching expected succeeds; stale expected conflicts.
|
||||
if err := s.CompareAndSwap(ctx, "user:1", []byte("v1"), []byte("v2")); err != nil {
|
||||
t.Fatalf("CompareAndSwap matching: %v", err)
|
||||
}
|
||||
if err := s.CompareAndSwap(ctx, "user:1", []byte("v1"), []byte("v3")); !errors.Is(err, ErrConflict) {
|
||||
t.Errorf("CompareAndSwap stale: got %v, want ErrConflict", err)
|
||||
}
|
||||
got, err := s.Get(ctx, "user:1")
|
||||
if err != nil {
|
||||
t.Fatalf("Get: %v", err)
|
||||
}
|
||||
if string(got) != "v2" {
|
||||
t.Errorf("stored value = %q, want %q", got, "v2")
|
||||
}
|
||||
|
||||
// Non-nil expected on a missing key conflicts (caller reloads and retries).
|
||||
if err := s.CompareAndSwap(ctx, "user:missing", []byte("v1"), []byte("v2")); !errors.Is(err, ErrConflict) {
|
||||
t.Errorf("CompareAndSwap missing key: got %v, want ErrConflict", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDynamoDBKVStore_DeleteMissingNoError(t *testing.T) {
|
||||
client, table, cleanup := dynamoDBLocalSetup(t)
|
||||
defer cleanup()
|
||||
|
||||
@@ -2,6 +2,7 @@ package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"reflect"
|
||||
"sort"
|
||||
@@ -61,6 +62,43 @@ func drainCollection(t *testing.T, c *firestore.Client, name string) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestFirestoreKV_CompareAndSwap(t *testing.T) {
|
||||
c := requireEmulator(t)
|
||||
col := uniqueCollection(t)
|
||||
defer drainCollection(t, c, col)
|
||||
store := NewFirestoreKVStore(c, col)
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
// Create-if-absent succeeds once, then conflicts.
|
||||
if err := store.CompareAndSwap(ctx, "user:1", nil, []byte("v1")); err != nil {
|
||||
t.Fatalf("CompareAndSwap create: %v", err)
|
||||
}
|
||||
if err := store.CompareAndSwap(ctx, "user:1", nil, []byte("v2")); !errors.Is(err, ErrConflict) {
|
||||
t.Errorf("CompareAndSwap create over existing: got %v, want ErrConflict", err)
|
||||
}
|
||||
|
||||
// Swap with matching expected succeeds; stale expected conflicts.
|
||||
if err := store.CompareAndSwap(ctx, "user:1", []byte("v1"), []byte("v2")); err != nil {
|
||||
t.Fatalf("CompareAndSwap matching: %v", err)
|
||||
}
|
||||
if err := store.CompareAndSwap(ctx, "user:1", []byte("v1"), []byte("v3")); !errors.Is(err, ErrConflict) {
|
||||
t.Errorf("CompareAndSwap stale: got %v, want ErrConflict", err)
|
||||
}
|
||||
got, err := store.Get(ctx, "user:1")
|
||||
if err != nil {
|
||||
t.Fatalf("Get: %v", err)
|
||||
}
|
||||
if string(got) != "v2" {
|
||||
t.Errorf("stored value = %q, want %q", got, "v2")
|
||||
}
|
||||
|
||||
// Non-nil expected on a missing key conflicts (caller reloads and retries).
|
||||
if err := store.CompareAndSwap(ctx, "user:missing", []byte("v1"), []byte("v2")); !errors.Is(err, ErrConflict) {
|
||||
t.Errorf("CompareAndSwap missing key: got %v, want ErrConflict", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFirestoreKV_PutGetRoundTrip(t *testing.T) {
|
||||
c := requireEmulator(t)
|
||||
col := uniqueCollection(t)
|
||||
|
||||
@@ -0,0 +1,91 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestMemoryKVStore_CompareAndSwap(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
existing []byte // nil = key absent before the call
|
||||
expected []byte
|
||||
val []byte
|
||||
wantErr error
|
||||
wantVal string // stored value after the call
|
||||
}{
|
||||
{
|
||||
name: "create when absent",
|
||||
existing: nil,
|
||||
expected: nil,
|
||||
val: []byte("v1"),
|
||||
wantErr: nil,
|
||||
wantVal: "v1",
|
||||
},
|
||||
{
|
||||
name: "create when exists conflicts",
|
||||
existing: []byte("v1"),
|
||||
expected: nil,
|
||||
val: []byte("v2"),
|
||||
wantErr: ErrConflict,
|
||||
wantVal: "v1",
|
||||
},
|
||||
{
|
||||
name: "swap when expected matches",
|
||||
existing: []byte("v1"),
|
||||
expected: []byte("v1"),
|
||||
val: []byte("v2"),
|
||||
wantErr: nil,
|
||||
wantVal: "v2",
|
||||
},
|
||||
{
|
||||
name: "swap when expected stale conflicts",
|
||||
existing: []byte("v2"),
|
||||
expected: []byte("v1"),
|
||||
val: []byte("v3"),
|
||||
wantErr: ErrConflict,
|
||||
wantVal: "v2",
|
||||
},
|
||||
{
|
||||
name: "swap when key missing conflicts",
|
||||
existing: nil,
|
||||
expected: []byte("v1"),
|
||||
val: []byte("v2"),
|
||||
wantErr: ErrConflict,
|
||||
wantVal: "",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
s := NewMemoryKVStore()
|
||||
if tt.existing != nil {
|
||||
if err := s.Put(ctx, "k", tt.existing); err != nil {
|
||||
t.Fatalf("Put: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
err := s.CompareAndSwap(ctx, "k", tt.expected, tt.val)
|
||||
if !errors.Is(err, tt.wantErr) {
|
||||
t.Fatalf("CompareAndSwap: got %v, want %v", err, tt.wantErr)
|
||||
}
|
||||
|
||||
got, err := s.Get(ctx, "k")
|
||||
if tt.wantVal == "" {
|
||||
if !errors.Is(err, ErrNotFound) {
|
||||
t.Fatalf("Get after failed create: got (%q, %v), want ErrNotFound", got, err)
|
||||
}
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatalf("Get: %v", err)
|
||||
}
|
||||
if string(got) != tt.wantVal {
|
||||
t.Errorf("stored value = %q, want %q", got, tt.wantVal)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -2,6 +2,8 @@ package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
)
|
||||
|
||||
@@ -37,7 +39,9 @@ func (p *prefixedStore) Put(ctx context.Context, key string, val []byte) error {
|
||||
func (p *prefixedStore) CompareAndSwap(ctx context.Context, key string, expected []byte, val []byte) error {
|
||||
cas, ok := p.inner.(CompareAndSwapStore)
|
||||
if !ok {
|
||||
return ErrConflict
|
||||
// A missing capability is permanent: report it as unsupported so
|
||||
// callers fail fast instead of retrying it as a write conflict.
|
||||
return fmt.Errorf("storage: inner store does not support compare-and-swap: %w", errors.ErrUnsupported)
|
||||
}
|
||||
return cas.CompareAndSwap(ctx, p.k(key), expected, val)
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"reflect"
|
||||
"testing"
|
||||
)
|
||||
@@ -67,6 +68,56 @@ func TestPrefixed_NotFoundPropagates(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestPrefixed_CompareAndSwapDelegatesWithPrefixedKey(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
base := NewMemoryKVStore()
|
||||
mod := Prefixed(base, "gold")
|
||||
|
||||
cas, ok := mod.(CompareAndSwapStore)
|
||||
if !ok {
|
||||
t.Fatal("Prefixed store does not implement CompareAndSwapStore")
|
||||
}
|
||||
if err := cas.CompareAndSwap(ctx, "user:1", nil, []byte("v1")); err != nil {
|
||||
t.Fatalf("CompareAndSwap create: %v", err)
|
||||
}
|
||||
|
||||
// The write must land under the prefixed key on the inner store.
|
||||
got, err := base.Get(ctx, "gold:user:1")
|
||||
if err != nil {
|
||||
t.Fatalf("inner Get: %v", err)
|
||||
}
|
||||
if string(got) != "v1" {
|
||||
t.Errorf("inner value = %q, want %q", got, "v1")
|
||||
}
|
||||
|
||||
// Stale expected must surface the inner store's conflict.
|
||||
if err := cas.CompareAndSwap(ctx, "user:1", []byte("stale"), []byte("v2")); !errors.Is(err, ErrConflict) {
|
||||
t.Errorf("CompareAndSwap stale: got %v, want ErrConflict", err)
|
||||
}
|
||||
if err := cas.CompareAndSwap(ctx, "user:1", []byte("v1"), []byte("v2")); err != nil {
|
||||
t.Errorf("CompareAndSwap matching: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// plainStore hides the inner store's CompareAndSwap method so the wrapper's
|
||||
// missing-capability path can be exercised.
|
||||
type plainStore struct {
|
||||
KVStore
|
||||
}
|
||||
|
||||
func TestPrefixed_CompareAndSwapUnsupportedInner(t *testing.T) {
|
||||
mod := Prefixed(&plainStore{KVStore: NewMemoryKVStore()}, "gold")
|
||||
cas := mod.(CompareAndSwapStore)
|
||||
|
||||
err := cas.CompareAndSwap(context.Background(), "user:1", nil, []byte("v1"))
|
||||
if !errors.Is(err, errors.ErrUnsupported) {
|
||||
t.Errorf("got %v, want errors.ErrUnsupported", err)
|
||||
}
|
||||
if errors.Is(err, ErrConflict) {
|
||||
t.Error("missing CAS capability must not be reported as a retryable conflict")
|
||||
}
|
||||
}
|
||||
|
||||
func TestPrefixed_PanicsOnEmptyPrefix(t *testing.T) {
|
||||
defer func() {
|
||||
if recover() == nil {
|
||||
|
||||
Reference in New Issue
Block a user