implement fix

CHANGES.txt

@@ -1,3 +1,6 @@
+5.4.1 (Oct 31, 2023)
+- Fix issue in split proxy where removed segment keys would be returned as active at startup
+
 5.4.0 (July 18, 2023)
 - Improved streaming architecture implementation to apply feature flag updates from the notification received which is now enhanced, improving efficiency and reliability of the whole update system.
 - Fixed possible edge case issue where deleting a feature flag doesn’t propagate immediately.

Makefile

@@ -59,6 +59,10 @@ test: $(sources) go.sum
 test_coverage: $(sources) go.sum
 	$(GO) test -v -cover -coverprofile=coverage.out $(ARGS) ./...
 
+## display unit test coverage derived from last test run (use `make test display-coverage` for up-to-date results)
+display-coverage: coverage.out
+	go tool cover -html=coverage.out
+
 ## Generate binaires for all architectures, ready to upload for distribution (with and without version)
 release_assets: \
     $(BUILD)/synchronizer \
@@ -166,6 +170,10 @@ table_header:
 	@echo "| **Command line option** | **JSON option** | **Environment variable** (container-only) | **Description** |"
 	@echo "| --- | --- | --- | --- |"
 
+coverage.out: test_coverage
+
+
+
 # Help target borrowed from: https://docs.cloudposse.com/reference/best-practices/make-best-practices/
 ## This help screen
 help:

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/splitio/gincache v1.0.1
 	github.com/splitio/go-split-commons/v5 v5.0.0
 	github.com/splitio/go-toolkit/v5 v5.3.1
+	github.com/stretchr/testify v1.8.3
 	go.etcd.io/bbolt v1.3.6
 )
 
@@ -19,6 +20,7 @@ require (
 	github.com/bytedance/sonic v1.9.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/gabriel-vasile/mimetype v1.4.2 // indirect
 	github.com/gin-contrib/sse v0.1.0 // indirect
@@ -34,7 +36,9 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pelletier/go-toml/v2 v2.0.8 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/redis/go-redis/v9 v9.0.4 // indirect
+	github.com/stretchr/objx v0.5.0 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	github.com/ugorji/go/codec v1.2.11 // indirect
 	golang.org/x/arch v0.3.0 // indirect

go.sum

@@ -93,6 +93,7 @@ github.com/splitio/go-toolkit/v5 v5.3.1 h1:9J/byd0fRxWj5/Zg0QZOnUxKBDIAMCGr7rySY
 github.com/splitio/go-toolkit/v5 v5.3.1/go.mod h1:xYhUvV1gga9/1029Wbp5pjnR6Cy8nvBpjw99wAbsMko=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/objx v0.5.0 h1:1zr/of2m5FGMsad5YfcqgdqdWrIhu+EBEJRhR1U7z/c=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=

splitio/commitversion.go

@@ -5,4 +5,4 @@ This file is created automatically, please do not edit
 */
 
 // CommitVersion is the version of the last commit previous to release
-const CommitVersion = "da63b9f"
+const CommitVersion = "fa204db"

splitio/proxy/storage/persistent/mocks/segment.go

@@ -0,0 +1,28 @@
+package mocks
+
+import (
+	"github.com/splitio/go-toolkit/v5/datastructures/set"
+	"github.com/splitio/split-synchronizer/v5/splitio/proxy/storage/persistent"
+	"github.com/stretchr/testify/mock"
+)
+
+type SegmentChangesCollectionMock struct {
+	mock.Mock
+}
+
+func (s *SegmentChangesCollectionMock) Update(name string, toAdd *set.ThreadUnsafeSet, toRemove *set.ThreadUnsafeSet, cn int64) error {
+	return s.Called(name, toAdd, toRemove, cn).Error(0)
+}
+
+func (s *SegmentChangesCollectionMock) Fetch(name string) (*persistent.SegmentChangesItem, error) {
+	args := s.Called(name)
+	return args.Get(0).(*persistent.SegmentChangesItem), args.Error(1)
+}
+
+func (s *SegmentChangesCollectionMock) ChangeNumber(segment string) int64 {
+	return s.Called(segment).Get(0).(int64)
+}
+
+func (s *SegmentChangesCollectionMock) SetChangeNumber(segment string, cn int64) {
+	s.Called(segment, cn)
+}

splitio/proxy/storage/persistent/segments.go

@@ -25,25 +25,32 @@ type SegmentChangesItem struct {
 	Keys map[string]SegmentKey
 }
 
-// SegmentChangesCollection represents a collection of SplitChangesItem
-type SegmentChangesCollection struct {
+type SegmentChangesCollection interface {
+	Update(name string, toAdd *set.ThreadUnsafeSet, toRemove *set.ThreadUnsafeSet, cn int64) error
+	Fetch(name string) (*SegmentChangesItem, error)
+	ChangeNumber(segment string) int64
+	SetChangeNumber(segment string, cn int64)
+}
+
+// SegmentChangesCollectionImpl represents a collection of SplitChangesItem
+type SegmentChangesCollectionImpl struct {
 	collection   CollectionWrapper
 	segmentsTill map[string]int64
 	logger       logging.LoggerInterface
 	mutex        sync.RWMutex
 }
 
 // NewSegmentChangesCollection returns an instance of SegmentChangesCollection
-func NewSegmentChangesCollection(db DBWrapper, logger logging.LoggerInterface) *SegmentChangesCollection {
-	return &SegmentChangesCollection{
+func NewSegmentChangesCollection(db DBWrapper, logger logging.LoggerInterface) *SegmentChangesCollectionImpl {
+	return &SegmentChangesCollectionImpl{
 		collection:   &BoltDBCollectionWrapper{db: db, name: segmentChangesCollectionName, logger: logger},
 		segmentsTill: make(map[string]int64, 0),
 		logger:       logger,
 	}
 }
 
 // Update persists a segmentChanges update
-func (c *SegmentChangesCollection) Update(name string, toAdd *set.ThreadUnsafeSet, toRemove *set.ThreadUnsafeSet, cn int64) error {
+func (c *SegmentChangesCollectionImpl) Update(name string, toAdd *set.ThreadUnsafeSet, toRemove *set.ThreadUnsafeSet, cn int64) error {
 	c.mutex.Lock()
 	defer c.mutex.Unlock()
 
@@ -63,17 +70,10 @@ func (c *SegmentChangesCollection) Update(name string, toAdd *set.ThreadUnsafeSe
 			continue
 		}
 		c.logger.Debug("Removing", strKey, "from", name)
-		if _, exists := segmentItem.Keys[strKey]; exists {
-			itemAux := segmentItem.Keys[strKey]
-			itemAux.Removed = true
-			itemAux.ChangeNumber = cn
-			segmentItem.Keys[strKey] = itemAux
-		} else {
-			segmentItem.Keys[strKey] = SegmentKey{
-				Name:         strKey,
-				Removed:      true,
-				ChangeNumber: cn,
-			}
+		segmentItem.Keys[strKey] = SegmentKey{
+			Name:         strKey,
+			Removed:      true,
+			ChangeNumber: cn,
 		}
 
 	}
@@ -85,17 +85,10 @@ func (c *SegmentChangesCollection) Update(name string, toAdd *set.ThreadUnsafeSe
 			continue
 		}
 		c.logger.Debug("Adding", strKey, "in", name)
-		if _, exists := segmentItem.Keys[strKey]; exists {
-			itemAux := segmentItem.Keys[strKey]
-			itemAux.Removed = false
-			itemAux.ChangeNumber = cn
-			segmentItem.Keys[strKey] = itemAux
-		} else {
-			segmentItem.Keys[strKey] = SegmentKey{
-				Name:         strKey,
-				Removed:      false,
-				ChangeNumber: cn,
-			}
+		segmentItem.Keys[strKey] = SegmentKey{
+			Name:         strKey,
+			Removed:      false,
+			ChangeNumber: cn,
 		}
 	}
 
@@ -108,13 +101,13 @@ func (c *SegmentChangesCollection) Update(name string, toAdd *set.ThreadUnsafeSe
 }
 
 // Fetch return a SegmentChangesItem
-func (c *SegmentChangesCollection) Fetch(name string) (*SegmentChangesItem, error) {
+func (c *SegmentChangesCollectionImpl) Fetch(name string) (*SegmentChangesItem, error) {
 	c.mutex.RLock()
 	defer c.mutex.RUnlock()
 	return c.fetch(name)
 }
 
-func (c *SegmentChangesCollection) fetch(name string) (*SegmentChangesItem, error) {
+func (c *SegmentChangesCollectionImpl) fetch(name string) (*SegmentChangesItem, error) {
 	item, err := c.collection.FetchBy([]byte(name))
 	if err != nil {
 		return nil, err
@@ -133,7 +126,7 @@ func (c *SegmentChangesCollection) fetch(name string) (*SegmentChangesItem, erro
 }
 
 // FetchAll return a list of SegmentChangesItem
-func (c *SegmentChangesCollection) FetchAll() ([]SegmentChangesItem, error) {
+func (c *SegmentChangesCollectionImpl) FetchAll() ([]SegmentChangesItem, error) {
 	c.mutex.RLock()
 	defer c.mutex.RUnlock()
 	items, err := c.collection.FetchAll()
@@ -163,7 +156,7 @@ func (c *SegmentChangesCollection) FetchAll() ([]SegmentChangesItem, error) {
 }
 
 // ChangeNumber returns changeNumber
-func (c *SegmentChangesCollection) ChangeNumber(segment string) int64 {
+func (c *SegmentChangesCollectionImpl) ChangeNumber(segment string) int64 {
 	c.mutex.RLock()
 	defer c.mutex.RUnlock()
 	value, exists := c.segmentsTill[segment]
@@ -174,8 +167,10 @@ func (c *SegmentChangesCollection) ChangeNumber(segment string) int64 {
 }
 
 // SetChangeNumber returns changeNumber
-func (c *SegmentChangesCollection) SetChangeNumber(segment string, cn int64) {
+func (c *SegmentChangesCollectionImpl) SetChangeNumber(segment string, cn int64) {
 	c.mutex.Lock()
 	defer c.mutex.Unlock()
 	c.segmentsTill[segment] = cn
 }
+
+var _ SegmentChangesCollection = (*SegmentChangesCollectionImpl)(nil)

splitio/proxy/storage/segments.go

@@ -29,7 +29,7 @@ type ProxySegmentStorage interface {
 type ProxySegmentStorageImpl struct {
 	logger         logging.LoggerInterface
 	nameCountCache *observability.ActiveSegmentTracker
-	db             *persistent.SegmentChangesCollection
+	db             persistent.SegmentChangesCollection
 	mysegments     optimized.MySegmentsCache
 }
 
@@ -68,21 +68,25 @@ func (s *ProxySegmentStorageImpl) ChangesSince(name string, since int64) (*dtos.
 
 	// Horrible loop borrowed from sdk-api
 	for _, skey := range item.Keys {
+
 		if skey.ChangeNumber <= since { // if the key was updated in a previous/current CN, we don't need to return it
 			continue
 		}
 
+		if skey.Removed && since < 0 {
+			// removed keys should not be returned on initialization payloads
+			continue
+		}
+
 		// Add the key to the corresponding list
-		if skey.Removed && since > 0 {
+		if skey.Removed {
 			removed = append(removed, skey.Name)
 		} else {
 			added = append(added, skey.Name)
 		}
 
 		// Update the till to be returned if necessary
-		if since > 0 && skey.ChangeNumber > till {
-			till = skey.ChangeNumber
-		} else if !skey.Removed && skey.ChangeNumber > till {
+		if skey.ChangeNumber > till {
 			till = skey.ChangeNumber
 		}
 	}
@@ -177,7 +181,7 @@ func (s *ProxySegmentStorageImpl) NamesAndCount() map[string]int {
 func populateCachesFromDisk(
 	dst optimized.MySegmentsCache,
 	names *observability.ActiveSegmentTracker,
-	src *persistent.SegmentChangesCollection,
+	src *persistent.SegmentChangesCollectionImpl,
 	logger logging.LoggerInterface,
 ) {
 	all, err := src.FetchAll()

splitio/proxy/storage/segments_test.go

@@ -0,0 +1,75 @@
+package storage
+
+import (
+	"testing"
+
+	"github.com/splitio/go-toolkit/v5/logging"
+	"github.com/splitio/split-synchronizer/v5/splitio/proxy/storage/optimized"
+	"github.com/splitio/split-synchronizer/v5/splitio/proxy/storage/persistent"
+	"github.com/splitio/split-synchronizer/v5/splitio/proxy/storage/persistent/mocks"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSegmentStorage(t *testing.T) {
+
+	psm := &mocks.SegmentChangesCollectionMock{}
+	psm.On("Fetch", "some").Return(&persistent.SegmentChangesItem{
+		Name: "some",
+		Keys: map[string]persistent.SegmentKey{
+			"k1": {Name: "k1", ChangeNumber: 1, Removed: false},
+			"k2": {Name: "k2", ChangeNumber: 1, Removed: true},
+			"k3": {Name: "k3", ChangeNumber: 2, Removed: false},
+			"k4": {Name: "k4", ChangeNumber: 2, Removed: true},
+			"k5": {Name: "k5", ChangeNumber: 3, Removed: false},
+			"k6": {Name: "k6", ChangeNumber: 3, Removed: true},
+			"k7": {Name: "k7", ChangeNumber: 4, Removed: false},
+		},
+	}, nil)
+
+	ss := ProxySegmentStorageImpl{
+		logger:     logging.NewLogger(nil),
+		db:         psm,
+		mysegments: optimized.NewMySegmentsCache(),
+	}
+
+	changes, err := ss.ChangesSince("some", -1)
+	assert.Nil(t, err)
+	assert.Equal(t, "some", changes.Name)
+	assert.ElementsMatch(t, []string{"k1", "k3", "k5", "k7"}, changes.Added)
+	assert.ElementsMatch(t, []string{}, changes.Removed)
+	assert.Equal(t, int64(-1), changes.Since)
+	assert.Equal(t, int64(4), changes.Till)
+
+	changes, err = ss.ChangesSince("some", 1)
+	assert.Nil(t, err)
+	assert.Equal(t, "some", changes.Name)
+	assert.ElementsMatch(t, []string{"k3", "k5", "k7"}, changes.Added)
+	assert.ElementsMatch(t, []string{"k4", "k6"}, changes.Removed)
+	assert.Equal(t, int64(1), changes.Since)
+	assert.Equal(t, int64(4), changes.Till)
+
+	changes, err = ss.ChangesSince("some", 2)
+	assert.Nil(t, err)
+	assert.Equal(t, "some", changes.Name)
+	assert.ElementsMatch(t, []string{"k5", "k7"}, changes.Added)
+	assert.ElementsMatch(t, []string{"k6"}, changes.Removed)
+	assert.Equal(t, int64(2), changes.Since)
+	assert.Equal(t, int64(4), changes.Till)
+
+	changes, err = ss.ChangesSince("some", 3)
+	assert.Nil(t, err)
+	assert.Equal(t, "some", changes.Name)
+	assert.ElementsMatch(t, []string{"k7"}, changes.Added)
+	assert.ElementsMatch(t, []string{}, changes.Removed)
+	assert.Equal(t, int64(3), changes.Since)
+	assert.Equal(t, int64(4), changes.Till)
+
+	changes, err = ss.ChangesSince("some", 4)
+	assert.Nil(t, err)
+	assert.Equal(t, "some", changes.Name)
+	assert.ElementsMatch(t, []string{}, changes.Added)
+	assert.ElementsMatch(t, []string{}, changes.Removed)
+	assert.Equal(t, int64(4), changes.Since)
+	assert.Equal(t, int64(4), changes.Till)
+
+}

splitio/version.go

@@ -2,4 +2,4 @@
 package splitio
 
 // Version is the version of this Agent
-const Version = "5.4.0"
+const Version = "5.4.1"