refactor(storage): replace value-bytes CAS with version-field optimistic lock

Replace content-addressed storage versioning (CAS) with explicit version fields
on documents. Versions are managed via optimistic locking pattern: Get returns
version, Put increments atomically and fails if version mismatch detected.
Callers (portfolio, lolschedule) updated to use GetVersioned/PutVersioned,
simplifying concurrency handling and eliminating CAS overhead.
This commit is contained in:
tiennm99 committed 2026-06-28 12:52:32 +07:00
1 parent ab7a597541
commit cd82abbb0c
14 files changed
+419 -320

No files matched your search

+10 -10
View File
@@ -62,13 +62,13 @@ func SavePortfolio(ctx context.Context, kv storage.KVStore, userID int64, p Port
}
func UpdatePortfolio(ctx context.Context, kv storage.KVStore, userID int64, now int64, mutate func(*Portfolio) error) (Portfolio, error) {
cas, ok := kv.(storage.CompareAndSwapStore)
vs, ok := kv.(storage.VersionedStore)
if !ok {
return Portfolio{}, fmt.Errorf("coin: storage does not support conditional portfolio updates")
return Portfolio{}, fmt.Errorf("coin: storage does not support versioned portfolio updates")
}
key := portfolioKey(userID)
for attempt := 0; attempt < portfolioUpdateAttempts; attempt++ {
p, expected, err := loadPortfolioForUpdate(ctx, kv, key, now)
p, version, err := loadPortfolioForUpdate(ctx, vs, key, now)
if err != nil {
return Portfolio{}, fmt.Errorf("coin: load portfolio %d: %w", userID, err)
}
@@ -80,7 +80,7 @@ func UpdatePortfolio(ctx context.Context, kv storage.KVStore, userID int64, now
if err != nil {
return Portfolio{}, fmt.Errorf("coin: save portfolio %d: json encode: %w", userID, err)
}
if err := cas.CompareAndSwap(ctx, key, expected, next); err == nil {
if err := vs.PutVersioned(ctx, key, version, next); err == nil {
return p, nil
} else if !errors.Is(err, storage.ErrConflict) {
return Portfolio{}, fmt.Errorf("coin: save portfolio %d: %w", userID, err)
@@ -89,13 +89,13 @@ func UpdatePortfolio(ctx context.Context, kv storage.KVStore, userID int64, now
return Portfolio{}, fmt.Errorf("coin: save portfolio %d: %w", userID, storage.ErrConflict)
}
func loadPortfolioForUpdate(ctx context.Context, kv storage.KVStore, key string, now int64) (Portfolio, []byte, error) {
raw, err := kv.Get(ctx, key)
func loadPortfolioForUpdate(ctx context.Context, vs storage.VersionedStore, key string, now int64) (Portfolio, int64, error) {
raw, version, err := vs.GetVersioned(ctx, key)
switch {
case err == nil:
var p Portfolio
if err := json.Unmarshal(raw, &p); err != nil {
return Portfolio{}, nil, fmt.Errorf("json decode: %w", err)
return Portfolio{}, 0, fmt.Errorf("json decode: %w", err)
}
p.normalize()
if p.Assets == nil {
@@ -104,11 +104,11 @@ func loadPortfolioForUpdate(ctx context.Context, kv storage.KVStore, key string,
if p.Meta.CreatedAt == 0 {
p.Meta.CreatedAt = now
}
return p, raw, nil
return p, version, nil
case errors.Is(err, storage.ErrNotFound):
return NewPortfolio(now), nil, nil
return NewPortfolio(now), 0, nil
default:
return Portfolio{}, nil, err
return Portfolio{}, 0, err
}
}
+6 -2
View File
@@ -72,7 +72,11 @@ type conflictOnceStore struct {
conflicted bool
}
func (s *conflictOnceStore) CompareAndSwap(ctx context.Context, key string, expected []byte, val []byte) error {
func (s *conflictOnceStore) GetVersioned(ctx context.Context, key string) ([]byte, int64, error) {
return s.KVStore.(storage.VersionedStore).GetVersioned(ctx, key)
}
func (s *conflictOnceStore) PutVersioned(ctx context.Context, key string, expectedVersion int64, val []byte) error {
if !s.conflicted {
s.conflicted = true
competing := NewPortfolio(1)
@@ -82,7 +86,7 @@ func (s *conflictOnceStore) CompareAndSwap(ctx context.Context, key string, expe
}
return storage.ErrConflict
}
return s.KVStore.(storage.CompareAndSwapStore).CompareAndSwap(ctx, key, expected, val)
return s.KVStore.(storage.VersionedStore).PutVersioned(ctx, key, expectedVersion, val)
}
func TestUpdatePortfolioRetriesAfterWriteConflict(t *testing.T) {
+10 -10
View File
@@ -59,13 +59,13 @@ func SavePortfolio(ctx context.Context, kv storage.KVStore, userID int64, p Port
}
func UpdatePortfolio(ctx context.Context, kv storage.KVStore, userID int64, now int64, mutate func(*Portfolio) error) (Portfolio, error) {
cas, ok := kv.(storage.CompareAndSwapStore)
vs, ok := kv.(storage.VersionedStore)
if !ok {
return Portfolio{}, fmt.Errorf("gold: storage does not support conditional portfolio updates")
return Portfolio{}, fmt.Errorf("gold: storage does not support versioned portfolio updates")
}
key := portfolioKey(userID)
for attempt := 0; attempt < portfolioUpdateAttempts; attempt++ {
p, expected, err := loadPortfolioForUpdate(ctx, kv, key, now)
p, version, err := loadPortfolioForUpdate(ctx, vs, key, now)
if err != nil {
return Portfolio{}, fmt.Errorf("gold: load portfolio %d: %w", userID, err)
}
@@ -77,7 +77,7 @@ func UpdatePortfolio(ctx context.Context, kv storage.KVStore, userID int64, now
if err != nil {
return Portfolio{}, fmt.Errorf("gold: save portfolio %d: json encode: %w", userID, err)
}
if err := cas.CompareAndSwap(ctx, key, expected, next); err == nil {
if err := vs.PutVersioned(ctx, key, version, next); err == nil {
return p, nil
} else if !errors.Is(err, storage.ErrConflict) {
return Portfolio{}, fmt.Errorf("gold: save portfolio %d: %w", userID, err)
@@ -86,23 +86,23 @@ func UpdatePortfolio(ctx context.Context, kv storage.KVStore, userID int64, now
return Portfolio{}, fmt.Errorf("gold: save portfolio %d: %w", userID, storage.ErrConflict)
}
func loadPortfolioForUpdate(ctx context.Context, kv storage.KVStore, key string, now int64) (Portfolio, []byte, error) {
raw, err := kv.Get(ctx, key)
func loadPortfolioForUpdate(ctx context.Context, vs storage.VersionedStore, key string, now int64) (Portfolio, int64, error) {
raw, version, err := vs.GetVersioned(ctx, key)
switch {
case err == nil:
var p Portfolio
if err := json.Unmarshal(raw, &p); err != nil {
return Portfolio{}, nil, fmt.Errorf("json decode: %w", err)
return Portfolio{}, 0, fmt.Errorf("json decode: %w", err)
}
p.normalize()
if p.Meta.CreatedAt == 0 {
p.Meta.CreatedAt = now
}
return p, raw, nil
return p, version, nil
case errors.Is(err, storage.ErrNotFound):
return NewPortfolio(now), nil, nil
return NewPortfolio(now), 0, nil
default:
return Portfolio{}, nil, err
return Portfolio{}, 0, err
}
}
+15 -7
View File
@@ -76,7 +76,11 @@ type conflictOnceStore struct {
conflicted bool
}
func (s *conflictOnceStore) CompareAndSwap(ctx context.Context, key string, expected []byte, val []byte) error {
func (s *conflictOnceStore) GetVersioned(ctx context.Context, key string) ([]byte, int64, error) {
return s.KVStore.(storage.VersionedStore).GetVersioned(ctx, key)
}
func (s *conflictOnceStore) PutVersioned(ctx context.Context, key string, expectedVersion int64, val []byte) error {
if !s.conflicted {
s.conflicted = true
competing := NewPortfolio(1)
@@ -86,7 +90,7 @@ func (s *conflictOnceStore) CompareAndSwap(ctx context.Context, key string, expe
}
return storage.ErrConflict
}
return s.KVStore.(storage.CompareAndSwapStore).CompareAndSwap(ctx, key, expected, val)
return s.KVStore.(storage.VersionedStore).PutVersioned(ctx, key, expectedVersion, val)
}
func TestUpdatePortfolioRetriesAfterWriteConflict(t *testing.T) {
@@ -116,7 +120,11 @@ type alwaysConflictStore struct {
attempts int
}
func (s *alwaysConflictStore) CompareAndSwap(context.Context, string, []byte, []byte) error {
func (s *alwaysConflictStore) GetVersioned(ctx context.Context, key string) ([]byte, int64, error) {
return s.KVStore.(storage.VersionedStore).GetVersioned(ctx, key)
}
func (s *alwaysConflictStore) PutVersioned(context.Context, string, int64, []byte) error {
s.attempts++
return storage.ErrConflict
}
@@ -144,9 +152,9 @@ type countingCASStore struct {
casCalls int
}
func (s *countingCASStore) CompareAndSwap(ctx context.Context, key string, expected, val []byte) error {
func (s *countingCASStore) PutVersioned(ctx context.Context, key string, expectedVersion int64, val []byte) error {
s.casCalls++
return s.MemoryKVStore.CompareAndSwap(ctx, key, expected, val)
return s.MemoryKVStore.PutVersioned(ctx, key, expectedVersion, val)
}
func TestUpdatePortfolioMutateErrorDoesNotRetryOrPersist(t *testing.T) {
@@ -212,8 +220,8 @@ func TestUpdatePortfolioConcurrentIncrementsLoseNoUpdates(t *testing.T) {
}
}
// plainKVStore hides the embedded store's CompareAndSwap method to model a
// backend without CAS support.
// plainKVStore hides the embedded store's versioned methods to model a backend
// without optimistic-locking support.
type plainKVStore struct {
storage.KVStore
}
+22 -20
View File
@@ -134,37 +134,39 @@ func (s *state) dailyPushHandler(ctx context.Context, deps modules.Deps) error {
// for the daily push: a winner proceeds to fan out; a loser (another trigger
// already claimed today) returns false and sends nothing.
//
// The claim is a compare-and-swap on lastPushDateKey so two simultaneous
// triggers cannot both win. Every real KV backend implements
// CompareAndSwapStore; if a bare store without CAS is used (only in narrow
// tests), it falls back to a plain Put — losing the atomic guarantee but
// preserving the "no-op if already today" behaviour.
// The claim is a version-based optimistic write on lastPushDateKey so two
// simultaneous triggers cannot both win. Every real KV backend implements
// VersionedStore; if a bare store without it is used (only in narrow tests), it
// falls back to a plain Put — losing the atomic guarantee but preserving the
// "no-op if already today" behaviour.
func claimDailyPush(ctx context.Context, kv storage.KVStore, today string) (bool, error) {
current, err := kv.Get(ctx, lastPushDateKey)
vs, ok := kv.(storage.VersionedStore)
if !ok {
// No versioned support (a bare store in narrow tests): best-effort Put,
// losing the atomic guarantee but preserving "no-op if already today".
current, err := kv.Get(ctx, lastPushDateKey)
if err == nil && string(current) == today {
return false, nil
}
if err != nil && !errors.Is(err, storage.ErrNotFound) {
return false, err
}
return true, kv.Put(ctx, lastPushDateKey, []byte(today))
}
current, version, err := vs.GetVersioned(ctx, lastPushDateKey)
switch {
case err == nil:
if string(current) == today {
return false, nil // already pushed today
}
case errors.Is(err, storage.ErrNotFound):
current = nil // never pushed
version = 0 // never pushed
default:
return false, err
}
cas, ok := kv.(storage.CompareAndSwapStore)
if !ok {
if err := kv.Put(ctx, lastPushDateKey, []byte(today)); err != nil {
return false, err
}
return true, nil
}
var expected []byte
if len(current) > 0 {
expected = current
}
if err := cas.CompareAndSwap(ctx, lastPushDateKey, expected, []byte(today)); err != nil {
if err := vs.PutVersioned(ctx, lastPushDateKey, version, []byte(today)); err != nil {
if errors.Is(err, storage.ErrConflict) {
return false, nil // another trigger claimed today first
}