From 3cad105a81d9f643b6dffbdab377199db55efdad Mon Sep 17 00:00:00 2001 From: tiennm99 Date: Thu, 11 Jun 2026 21:40:16 +0700 Subject: [PATCH] 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 --- internal/modules/gold/portfolio_test.go | 133 ++++++++++++++++++++++++ internal/storage/dynamodb_kv_test.go | 36 +++++++ internal/storage/firestore_kv_test.go | 38 +++++++ internal/storage/memory_kv_test.go | 91 ++++++++++++++++ internal/storage/prefix.go | 6 +- internal/storage/prefix_test.go | 51 +++++++++ 6 files changed, 354 insertions(+), 1 deletion(-) create mode 100644 internal/storage/memory_kv_test.go diff --git a/internal/modules/gold/portfolio_test.go b/internal/modules/gold/portfolio_test.go index f1a8875..a521e35 100644 --- a/internal/modules/gold/portfolio_test.go +++ b/internal/modules/gold/portfolio_test.go @@ -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() diff --git a/internal/storage/dynamodb_kv_test.go b/internal/storage/dynamodb_kv_test.go index 16e9778..bca9b0a 100644 --- a/internal/storage/dynamodb_kv_test.go +++ b/internal/storage/dynamodb_kv_test.go @@ -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() diff --git a/internal/storage/firestore_kv_test.go b/internal/storage/firestore_kv_test.go index 79e7694..6bd64e0 100644 --- a/internal/storage/firestore_kv_test.go +++ b/internal/storage/firestore_kv_test.go @@ -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) diff --git a/internal/storage/memory_kv_test.go b/internal/storage/memory_kv_test.go new file mode 100644 index 0000000..8b5d5c3 --- /dev/null +++ b/internal/storage/memory_kv_test.go @@ -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) + } + }) + } +} diff --git a/internal/storage/prefix.go b/internal/storage/prefix.go index 56ca00f..337641d 100644 --- a/internal/storage/prefix.go +++ b/internal/storage/prefix.go @@ -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) } diff --git a/internal/storage/prefix_test.go b/internal/storage/prefix_test.go index d0473b2..5bfc2df 100644 --- a/internal/storage/prefix_test.go +++ b/internal/storage/prefix_test.go @@ -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 {