add SetPersist

Author: cwaldren-ld

internal/datasystem/store.go

@@ -45,7 +45,7 @@ type Store struct {
 	// Points to the active store. Swapped upon initialization.
 	active subsystems.DataStore
 
-	quality DataQuality
+	persist bool
 
 	// Protects the availability, persistentStore, quality, and active fields.
 	mu sync.RWMutex
@@ -79,13 +79,19 @@ func NewStore(loggers ldlog.Loggers) *Store {
 	s := &Store{
 		persistentStore: nil,
 		memoryStore:     datastore.NewInMemoryDataStore(loggers),
-		quality:         QualityNone,
+		persist:         false,
 		loggers:         loggers,
 	}
 	s.active = s.memoryStore
 	return s
 }
 
+func (s *Store) SetPersist(persist bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.persist = persist
+}
+
 // Close closes the store. If there is a persistent store configured, it will be closed.
 func (s *Store) Close() error {
 	s.mu.Lock()
@@ -105,12 +111,11 @@ func (s *Store) getActive() subsystems.DataStore {
 }
 
 // Mirroring returns true data is being mirrored to a persistent store.
-func (s *Store) mirroring() bool {
-	return s.persistentStore != nil && s.persistentStore.mode == subsystems.DataStoreModeReadWrite &&
-		s.quality == QualityTrusted
+func (s *Store) shouldPersist() bool {
+	return s.persist && s.persistentStore != nil && s.persistentStore.mode == subsystems.DataStoreModeReadWrite
 }
 
-// nolint:revive // Standard DataSourceUpdateSink method
+// nolint:revive // Standard DataDestination method
 func (s *Store) Init(allData []ldstoretypes.Collection, payloadVersion *int) bool {
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -119,16 +124,15 @@ func (s *Store) Init(allData []ldstoretypes.Collection, payloadVersion *int) boo
 	// TODO: handle errors from initializing the memory or persistent stores.
 	if err := s.memoryStore.Init(allData); err == nil {
 		s.active = s.memoryStore
-		s.quality = QualityTrusted
 	}
 
-	if s.mirroring() {
+	if s.shouldPersist() {
 		_ = s.persistentStore.impl.Init(allData) // TODO: insert in topo-sort order
 	}
 	return true
 }
 
-// nolint:revive // Standard DataSourceUpdateSink method
+// nolint:revive // Standard DataDestination method
 func (s *Store) Upsert(kind ldstoretypes.DataKind, key string, item ldstoretypes.ItemDescriptor) bool {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
@@ -141,13 +145,14 @@ func (s *Store) Upsert(kind ldstoretypes.DataKind, key string, item ldstoretypes
 	// TXNS-PS: Requirement 1.3.3, must apply updates to in-memory before the persistent store.
 	_, memErr = s.memoryStore.Upsert(kind, key, item)
 
-	if s.mirroring() {
+	if s.shouldPersist() {
 		_, persErr = s.persistentStore.impl.Upsert(kind, key, item)
 	}
 	return memErr == nil && persErr == nil
 }
 
-// nolint:revive // Standard DataSourceUpdateSink method
+// GetDataStoreStatusProvider returns the status provider for the persistent store, if one is configured, otherwise
+// nil.
 func (s *Store) GetDataStoreStatusProvider() interfaces.DataStoreStatusProvider {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
@@ -171,15 +176,14 @@ func (s *Store) WithPersistence(persistent subsystems.DataStore, mode subsystems
 	}
 
 	s.active = s.persistentStore.impl
-	s.quality = QualityUntrusted
 	return s
 }
 
 func (s *Store) Commit() error {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
 
-	if s.mirroring() {
+	if s.shouldPersist() {
 		flags, err := s.memoryStore.GetAll(datakinds.Features)
 		if err != nil {
 			return err
@@ -207,17 +211,3 @@ func (s *Store) Get(kind ldstoretypes.DataKind, key string) (ldstoretypes.ItemDe
 func (s *Store) IsInitialized() bool {
 	return s.getActive().IsInitialized()
 }
-
-type DataQuality int
-
-const (
-	QualityNone      = DataQuality(0)
-	QualityUntrusted = DataQuality(1)
-	QualityTrusted   = DataQuality(2)
-)
-
-func (s *Store) DataQuality() DataQuality {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-	return s.quality
-}

internal/datasystem/store_test.go

@@ -22,42 +22,29 @@ func TestStore_New(t *testing.T) {
 	assert.NoError(t, store.Close())
 }
 
-func TestStore_NoPersistence_NewStore_DataQuality(t *testing.T) {
-	logCapture := ldlogtest.NewMockLog()
-	store := NewStore(logCapture.Loggers)
-	defer store.Close()
-	assert.Equal(t, store.DataQuality(), QualityNone)
-}
-
 func TestStore_NoPersistence_NewStore_IsInitialized(t *testing.T) {
 	logCapture := ldlogtest.NewMockLog()
 	store := NewStore(logCapture.Loggers)
 	defer store.Close()
 	assert.False(t, store.IsInitialized())
 }
 
-func TestStore_NoPersistence_MemoryStoreInitialized_DataQualityIsTrusted(t *testing.T) {
-	// It doesn't matter if the data has a payload version or not: the data quality should be
-	// trusted if it came from an initializer or synchronizer.
-	// This isn't necessarily what we want going forward, the quality should vary depending on the
-	// initializer/synchronizer implementation.
+func TestStore_NoPersistence_MemoryStore_IsInitialized(t *testing.T) {
 
 	version1 := 1
 	tests := []struct {
 		name           string
 		payloadVersion *int
-		quality        DataQuality
 	}{
-		{"fresh data", &version1, QualityTrusted},
-		{"stale data", nil, QualityTrusted},
+		{"versioned data", &version1},
+		{"unversioned data", nil},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			logCapture := ldlogtest.NewMockLog()
 			store := NewStore(logCapture.Loggers)
 			defer store.Close()
 			store.Init([]ldstoretypes.Collection{}, tt.payloadVersion)
-			assert.Equal(t, store.DataQuality(), tt.quality)
 			assert.True(t, store.IsInitialized())
 		})
 	}
@@ -95,6 +82,8 @@ func TestStore_Commit(t *testing.T) {
 
 		spy.isDown = false
 
+		store.SetPersist(true)
+
 		require.NoError(t, store.Commit())
 
 		assert.Equal(t, initPayload, spy.initPayload)
@@ -119,6 +108,8 @@ func TestStore_Commit(t *testing.T) {
 
 		require.Empty(t, spy.initPayload)
 
+		store.SetPersist(true)
+
 		require.NoError(t, store.Commit())
 
 		assert.Empty(t, spy.initPayload)
@@ -144,6 +135,8 @@ func TestStore_Commit(t *testing.T) {
 
 		require.Empty(t, spy.initPayload)
 
+		store.SetPersist(true)
+
 		require.NoError(t, store.Commit())
 
 		assert.Empty(t, spy.initPayload)
@@ -212,7 +205,7 @@ func TestStore_Concurrency(t *testing.T) {
 			wg.Add(1)
 			defer wg.Done()
 			for i := 0; i < 100; i++ {
-				_ = store.DataQuality()
+				store.SetPersist(true)
 				time.Sleep(time.Duration(rand.Intn(10)) * time.Millisecond)
 			}
 		}()

subsystems/data_destination.go

@@ -11,6 +11,7 @@ import (
 // Do not use it.
 // You have been warned.
 type DataDestination interface {
+
 	// Init overwrites the current contents of the data store with a set of items for each collection.
 	//
 	// If the underlying data store returns an error during this operation, the SDK will log it,