fix(hazelcast): use atomic operations for tag list updates

store/hazelcast/hazelcast.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"slices"
 	"strings"
 	"time"
 
@@ -17,6 +18,9 @@ type HazelcastMapInterface interface {
 	Get(ctx context.Context, key any) (any, error)
 	GetEntryView(ctx context.Context, key any) (*types.SimpleEntryView, error)
 	SetWithTTL(ctx context.Context, key any, value any, ttl time.Duration) error
+	SetTTL(ctx context.Context, key any, ttl time.Duration) error
+	PutIfAbsentWithTTL(ctx context.Context, key any, value any, ttl time.Duration) (any, error)
+	ReplaceIfSame(ctx context.Context, key any, oldValue any, newValue any) (bool, error)
 	Remove(ctx context.Context, key any) (any, error)
 	Clear(ctx context.Context) error
 }
@@ -85,29 +89,59 @@ func (s *HazelcastStore) setTags(ctx context.Context, hzMap HazelcastMapInterfac
 		currentTag := tag
 		group.Go(func() error {
 			tagKey := fmt.Sprintf(HazelcastTagPattern, currentTag)
-			tagValue, err := hzMap.Get(ctx, tagKey)
-			if err != nil {
-				return err
-			}
 
-			newTagValue := key.(string)
-			if tagValue != nil {
-				cacheKeys := strings.Split(tagValue.(string), ",")
-				for _, cacheKey := range cacheKeys {
-					if key == cacheKey {
-						return nil
-					}
+			var err error
+			for i := 0; i < 3; i++ {
+				if err = s.addKeyToTagValue(ctx, hzMap, tagKey, key, ttl); err == nil {
+					return nil
 				}
-				cacheKeys = append(cacheKeys, key.(string))
-				newTagValue = strings.Join(cacheKeys, ",")
+				// loop to retry any failure (including race conditions)
 			}
-
-			return hzMap.SetWithTTL(ctx, tagKey, newTagValue, ttl)
+			return err
 		})
 	}
 	group.Wait()
 }
 
+func (s *HazelcastStore) addKeyToTagValue(ctx context.Context, hzMap HazelcastMapInterface, tagKey string, key any, ttl time.Duration) error {
+	tagValue, err := hzMap.Get(ctx, tagKey)
+	if err != nil {
+		return err
+	}
+
+	if tagValue == nil {
+		// first writer: try to insert atomically
+		prev, err := hzMap.PutIfAbsentWithTTL(ctx, tagKey, key.(string), ttl)
+		if err != nil {
+			return err
+		}
+		if prev == nil {
+			// PutIfAbsent returns the existing value or nil if absent;
+			// nil here means our insert succeeded.
+			return nil
+		}
+		// somebody else inserted; fall through to update path with the new value
+		tagValue = prev
+	}
+
+	oldStr := tagValue.(string)
+	cacheKeys := strings.Split(oldStr, ",")
+	if slices.Contains(cacheKeys, key.(string)) {
+		return hzMap.SetTTL(ctx, tagKey, ttl)
+	}
+	newStr := strings.Join(append(cacheKeys, key.(string)), ",")
+
+	ok, err := hzMap.ReplaceIfSame(ctx, tagKey, oldStr, newStr)
+	if err != nil {
+		return err
+	}
+	if !ok {
+		return errors.New("hazelcast tag key contended")
+	}
+
+	return hzMap.SetTTL(ctx, tagKey, ttl)
+}
+
 // Delete removes data from Hazelcast for given key identifier
 func (s *HazelcastStore) Delete(ctx context.Context, key any) error {
 	_, err := s.hzMap.Remove(ctx, key)

store/hazelcast/hazelcast_test.go

@@ -98,8 +98,8 @@ func TestHazelcastSetWithTags(t *testing.T) {
 
 	hzMap := NewMockHazelcastMapInterface(ctrl)
 	hzMap.EXPECT().SetWithTTL(ctx, cacheKey, cacheValue, time.Duration(0)).Return(nil)
-	hzMap.EXPECT().SetWithTTL(gomock.Any(), "gocache_tag_tag1", cacheKey, time.Duration(0)).Return(nil)
 	hzMap.EXPECT().Get(gomock.Any(), "gocache_tag_tag1").Return(nil, nil)
+	hzMap.EXPECT().PutIfAbsentWithTTL(gomock.Any(), "gocache_tag_tag1", cacheKey, time.Duration(0)).Return(nil, nil)
 
 	store := NewHazelcast(hzMap)
 
@@ -110,7 +110,7 @@ func TestHazelcastSetWithTags(t *testing.T) {
 	assert.Nil(t, err)
 }
 
-func TestHazelcastSetWithTagsTTL(t *testing.T) {
+func TestHazelcastSetWithTagsAndTagsTTL(t *testing.T) {
 	// Given
 	ctrl := gomock.NewController(t)
 
@@ -121,8 +121,8 @@ func TestHazelcastSetWithTagsTTL(t *testing.T) {
 
 	hzMap := NewMockHazelcastMapInterface(ctrl)
 	hzMap.EXPECT().SetWithTTL(ctx, cacheKey, cacheValue, time.Duration(0)).Return(nil)
-	hzMap.EXPECT().SetWithTTL(gomock.Any(), "gocache_tag_tag1", cacheKey, 10*time.Second).Return(nil)
 	hzMap.EXPECT().Get(gomock.Any(), "gocache_tag_tag1").Return(nil, nil)
+	hzMap.EXPECT().PutIfAbsentWithTTL(gomock.Any(), "gocache_tag_tag1", cacheKey, 10*time.Second).Return(nil, nil)
 
 	store := NewHazelcast(hzMap)
 
@@ -133,6 +133,53 @@ func TestHazelcastSetWithTagsTTL(t *testing.T) {
 	assert.Nil(t, err)
 }
 
+func TestHazelcastSetWithTagsWhenAlreadyInserted(t *testing.T) {
+	// Given
+	ctrl := gomock.NewController(t)
+
+	ctx := context.Background()
+
+	cacheKey := "my-key"
+	cacheValue := "my-cache-value"
+
+	hzMap := NewMockHazelcastMapInterface(ctrl)
+	hzMap.EXPECT().SetWithTTL(ctx, cacheKey, cacheValue, time.Duration(0)).Return(nil)
+	hzMap.EXPECT().Get(gomock.Any(), "gocache_tag_tag1").Return("my-key,a-second-key", nil)
+	hzMap.EXPECT().SetTTL(gomock.Any(), "gocache_tag_tag1", time.Duration(0)).Return(nil)
+
+	store := NewHazelcast(hzMap)
+
+	// When
+	err := store.Set(ctx, cacheKey, cacheValue, lib_store.WithTags([]string{"tag1"}))
+
+	// Then
+	assert.Nil(t, err)
+}
+
+func TestHazelcastSetWithTagsWhenTagExists(t *testing.T) {
+	// Given
+	ctrl := gomock.NewController(t)
+
+	ctx := context.Background()
+
+	cacheKey := "my-key"
+	cacheValue := "my-cache-value"
+
+	hzMap := NewMockHazelcastMapInterface(ctrl)
+	hzMap.EXPECT().SetWithTTL(ctx, cacheKey, cacheValue, time.Duration(0)).Return(nil)
+	hzMap.EXPECT().Get(gomock.Any(), "gocache_tag_tag1").Return("a-second-key", nil)
+	hzMap.EXPECT().ReplaceIfSame(gomock.Any(), "gocache_tag_tag1", "a-second-key", "a-second-key,my-key").Return(true, nil)
+	hzMap.EXPECT().SetTTL(gomock.Any(), "gocache_tag_tag1", time.Duration(0)).Return(nil)
+
+	store := NewHazelcast(hzMap)
+
+	// When
+	err := store.Set(ctx, cacheKey, cacheValue, lib_store.WithTags([]string{"tag1"}))
+
+	// Then
+	assert.Nil(t, err)
+}
+
 func TestHazelcastDelete(t *testing.T) {
 	// Given
 	ctrl := gomock.NewController(t)