fix: align fdv2 intializer to spec (#330)

**Requirements**

- [x] I have added test coverage for new or changed functionality
- [x] I have followed the repository's [pull request submission
guidelines](../blob/v5/CONTRIBUTING.md#submitting-pull-requests)
- [x] I have validated my changes against all supported platform
versions

**Describe the solution you've provided**

This PR will make it so that the initializers will keep cycling to the
first intializer to report a valid basis + selector OR if there are no
initializers left. In this case, the synchronizers will start. In this
new flow, synchronizers could put the client into a valid state.

Initialization phase will no longer throw an error unless the
initializers timeout.

**Describe alternatives you've considered**

We may want to introduce an option for customers to enable a more strict
initialization flow where the client will terminate if all initializers
fail, but that would need an addendum to the specs and a new PR.

**Additional context**

Add any other context about the pull request here.

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> FDv2 now only marks ready when an initializer provides a selector,
updates full-dataset change detection to consider version increases,
makes file data source emit NoSelector, and updates/extends tests
including a new e2e deferral case.
> 
> - **FDv2 core**:
> - Initializers now mark ready only if
`basis.ChangeSet.Selector().IsDefined()`; otherwise continue to
synchronizers.
> - **Store**:
> - `computeChangedItemsForFullDataSet` now flags affected items when
`oldItem.Version < newItem.Version` in addition to add/remove.
> - **File data source (v2)**:
> - Change sets are finished with `subsystems.NoSelector()`; remove
selector/version usage.
> - **Tests**:
>   - Update file data source tests to expect `NoSelector`.
> - Add `TestFDV2FileInitializerWillDeferToFirstSynchronizer` end-to-end
test validating deferral to the first synchronizer.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
e37a5a1. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: joker23

internal/datasystem/fdv2_datasystem.go

@@ -251,12 +251,14 @@ func (f *FDv2) runInitializers(ctx context.Context, closeWhenReady chan struct{}
 			continue
 		}
 		f.environmentIDProvider.SetEnvironmentID(basis.EnvironmentID)
-		f.loggers.Infof("Initialized via %s", initializer.Name())
 		f.store.Apply(basis.ChangeSet, basis.Persist)
-		f.readyOnce.Do(func() {
-			close(closeWhenReady)
-		})
-		return
+		if basis.ChangeSet.Selector().IsDefined() {
+			f.loggers.Infof("Initialized via %s", initializer.Name())
+			f.readyOnce.Do(func() {
+				close(closeWhenReady)
+			})
+			return
+		}
 	}
 }
 

internal/datasystem/store.go

@@ -399,10 +399,11 @@ func (s *Store) computeChangedItemsForFullDataSet(
 			}
 		}
 		for _, key := range allKeys {
-			_, haveOld := oldItems[key]
-			_, haveNew := newItems[key]
+			oldItem, haveOld := oldItems[key]
+			newItem, haveNew := newItems[key]
+
 			if haveOld || haveNew {
-				if !haveOld || !haveNew {
+				if !haveOld || !haveNew || oldItem.Version < newItem.Version {
 					s.dependencyTracker.addAffectedItems(affectedItems, toposort.NewVertex(kind, key))
 				}
 			}

ldclient_end_to_end_fdv2_test.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/launchdarkly/go-sdk-common/v3/ldlog"
 	"github.com/launchdarkly/go-server-sdk/v7/internal/sharedtest"
+	"github.com/launchdarkly/go-server-sdk/v7/ldfiledatav2"
 	"github.com/launchdarkly/go-server-sdk/v7/subsystems"
 	"github.com/launchdarkly/go-server-sdk/v7/testhelpers/ldservicesv2"
 
@@ -17,6 +18,7 @@ import (
 	"github.com/launchdarkly/go-server-sdk/v7/ldcomponents"
 	"github.com/launchdarkly/go-server-sdk/v7/testhelpers/ldservices"
 
+	testHelpers "github.com/launchdarkly/go-test-helpers/v3"
 	"github.com/launchdarkly/go-test-helpers/v3/httphelpers"
 
 	"github.com/stretchr/testify/assert"
@@ -354,3 +356,49 @@ func TestFDV2StreamingSynchronizerTimesOut(t *testing.T) {
 		assert.Len(t, logCapture.GetOutput(ldlog.Error), 0)
 	})
 }
+
+func TestFDV2FileInitializerWillDeferToFirstSynchronizer(t *testing.T) {
+	data := ldservicesv2.NewServerSDKData().Flags(alwaysTrueFlag)
+
+	protocol := ldservicesv2.NewStreamingProtocol().
+		WithIntent(subsystems.ServerIntent{Payload: subsystems.Payload{
+			ID: "fake-id", Target: 0, Code: subsystems.IntentTransferFull, Reason: "payload-missing",
+		}}).
+		WithPutObjects(data.ToPutObjects()).
+		WithTransferred("state", 1)
+
+	streamHandler, _ := ldservices.ServerSideStreamingV2ServiceProtocolHandler(protocol)
+
+	handler, requestsCh := httphelpers.RecordingHandler(streamHandler)
+
+	testHelpers.WithTempFileData([]byte(`{"flags": {"`+alwaysFalseFlag.Key+`": {"on": false}}, "segments": {}}`), func(filename string) {
+
+		httphelpers.WithServer(handler, func(server *httptest.Server) {
+			logCapture := ldlogtest.NewMockLog()
+
+			config := Config{
+				Logging: ldcomponents.Logging().Loggers(logCapture.Loggers),
+				DataSystem: ldcomponents.DataSystem().Custom().
+					Initializers(
+						ldfiledatav2.DataSource().FilePaths(filename).AsInitializer(),
+					).
+					Synchronizers(
+						ldcomponents.StreamingDataSourceV2().BaseURI(server.URL),
+						nil,
+					),
+			}
+
+			client, err := MakeCustomClient(testSdkKey, config, time.Second*5)
+
+			<-requestsCh
+
+			require.NoError(t, err)
+			defer client.Close()
+
+			assert.Equal(t, string(interfaces.DataSourceStateValid), string(client.GetDataSourceStatusProvider().GetStatus().State))
+
+			value, _ := client.BoolVariation(alwaysTrueFlag.Key, testUser, false)
+			assert.True(t, value)
+		})
+	})
+}

ldfiledatav2/file_data_source_impl.go

@@ -7,7 +7,6 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
-	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -29,7 +28,9 @@ import (
 type fileDataSource struct {
 	changeSetBroadcaster *internal.Broadcaster[subsystems.ChangeSet]
 	statusBroadcaster    *internal.Broadcaster[interfaces.DataSynchronizerStatus]
-	version              int
+	// NOTE: this is not really used anymore because file data sources at this
+	// moment will not report a selector.
+	version int
 
 	absFilePaths          []string
 	duplicateKeysHandling DuplicateKeysHandling
@@ -349,7 +350,11 @@ func mergeFileData(
 		}
 	}
 
-	return builder.Finish(subsystems.NewSelector(strconv.Itoa(version), version))
+	// File data source will not have a selector for now.
+	// NOTE: If we start supporting FDv2 data from file then this statement might change.
+	// When that happens we will construct the selector the same way that we construct it
+	// in the FDv2 polling data source.
+	return builder.Finish(subsystems.NoSelector())
 }
 
 func makeFlagWithValue(key string, v interface{}) *ldmodel.FeatureFlag {

ldfiledatav2/file_data_source_test.go

@@ -34,7 +34,7 @@ func TestSuccessfullyLoadsJsonFlags(t *testing.T) {
 		assert.Contains(t, keys, "my-flag")
 		assert.Contains(t, keys, "my-segment")
 
-		assert.Equal(t, 1, result.ChangeSet.Selector().Version())
+		assert.Equal(t, subsystems.NoSelector(), result.ChangeSet.Selector())
 		assert.Equal(t, subsystems.IntentTransferFull, result.ChangeSet.IntentCode())
 		assert.Equal(t, interfaces.DataSourceStateValid, result.State)
 		assert.Equal(t, interfaces.DataSourceErrorInfo{}, result.Error)
@@ -54,7 +54,7 @@ func TestSuccessfullyLoadsJsonFlagValues(t *testing.T) {
 		assert.NotNil(t, result.ChangeSet)
 		assert.Len(t, result.ChangeSet.Changes(), 1)
 		assert.Equal(t, "my-flag", result.ChangeSet.Changes()[0].Key)
-		assert.Equal(t, 1, result.ChangeSet.Selector().Version())
+		assert.Equal(t, subsystems.NoSelector(), result.ChangeSet.Selector())
 		assert.Equal(t, subsystems.IntentTransferFull, result.ChangeSet.IntentCode())
 		assert.Equal(t, interfaces.DataSourceStateValid, result.State)
 		assert.Equal(t, interfaces.DataSourceErrorInfo{}, result.Error)
@@ -80,7 +80,7 @@ flags:
 		assert.NotNil(t, result.ChangeSet)
 		assert.Len(t, result.ChangeSet.Changes(), 1)
 		assert.Equal(t, "my-flag", result.ChangeSet.Changes()[0].Key)
-		assert.Equal(t, 1, result.ChangeSet.Selector().Version())
+		assert.Equal(t, subsystems.NoSelector(), result.ChangeSet.Selector())
 		assert.Equal(t, subsystems.IntentTransferFull, result.ChangeSet.IntentCode())
 		assert.Equal(t, interfaces.DataSourceStateValid, result.State)
 		assert.Equal(t, interfaces.DataSourceErrorInfo{}, result.Error)
@@ -105,7 +105,7 @@ flagValues:
 		assert.NotNil(t, result.ChangeSet)
 		assert.Len(t, result.ChangeSet.Changes(), 1)
 		assert.Equal(t, "my-flag", result.ChangeSet.Changes()[0].Key)
-		assert.Equal(t, 1, result.ChangeSet.Selector().Version())
+		assert.Equal(t, subsystems.NoSelector(), result.ChangeSet.Selector())
 		assert.Equal(t, subsystems.IntentTransferFull, result.ChangeSet.IntentCode())
 		assert.Equal(t, interfaces.DataSourceStateValid, result.State)
 		assert.Equal(t, interfaces.DataSourceErrorInfo{}, result.Error)
@@ -133,7 +133,7 @@ func TestSuccessfullyLoadsJsonFlagsFromMultipleFiles(t *testing.T) {
 			assert.Contains(t, keys, "my-flag")
 			assert.Contains(t, keys, "your-flag")
 
-			assert.Equal(t, 1, result.ChangeSet.Selector().Version())
+			assert.Equal(t, subsystems.NoSelector(), result.ChangeSet.Selector())
 			assert.Equal(t, subsystems.IntentTransferFull, result.ChangeSet.IntentCode())
 			assert.Equal(t, interfaces.DataSourceStateValid, result.State)
 			assert.Equal(t, interfaces.DataSourceErrorInfo{}, result.Error)
@@ -183,7 +183,7 @@ func TestIsValidWhenLoadingConflictingFilesWhenIgnoringDuplicates(t *testing.T)
 			// It uses the first duplicate key.
 			assert.Equal(t, 1, keys["my-flag"].Version)
 
-			assert.Equal(t, 1, result.ChangeSet.Selector().Version())
+			assert.Equal(t, subsystems.NoSelector(), result.ChangeSet.Selector())
 			assert.Equal(t, subsystems.IntentTransferFull, result.ChangeSet.IntentCode())
 			assert.Equal(t, interfaces.DataSourceStateValid, result.State)
 			assert.Equal(t, interfaces.DataSourceErrorInfo{}, result.Error)
@@ -238,7 +238,7 @@ func TestRecoversFromInvalidDataWhenFileUpdated(t *testing.T) {
 			assert.NotNil(t, result.ChangeSet)
 			assert.Len(t, result.ChangeSet.Changes(), 1)
 			assert.Equal(t, "my-flag", result.ChangeSet.Changes()[0].Key)
-			assert.Equal(t, 1, result.ChangeSet.Selector().Version())
+			assert.Equal(t, subsystems.NoSelector(), result.ChangeSet.Selector())
 			assert.Equal(t, subsystems.IntentTransferFull, result.ChangeSet.IntentCode())
 			assert.Equal(t, interfaces.DataSourceStateValid, result.State)
 		})
@@ -297,7 +297,7 @@ func TestSuccessfullyLoadsJsonFlagsAsInitializer(t *testing.T) {
 		assert.Contains(t, keys, "my-flag")
 		assert.Contains(t, keys, "my-segment")
 
-		assert.Equal(t, 1, basis.ChangeSet.Selector().Version())
+		assert.Equal(t, subsystems.NoSelector(), basis.ChangeSet.Selector())
 		assert.Equal(t, subsystems.IntentTransferFull, basis.ChangeSet.IntentCode())
 	})
 }