fix: Honor x-ld-fd-fallback header in fdv2 initializer phase (SDK-2203)

internal/datasourcev2/helpers.go

@@ -12,6 +12,16 @@ import (
 	"github.com/launchdarkly/go-sdk-common/v3/ldlog"
 )
 
+// fdv1FallbackHeader is the response header LaunchDarkly uses to instruct an SDK to abandon the
+// FDv2 protocol and fall back to FDv1.
+const fdv1FallbackHeader = "X-LD-FD-Fallback"
+
+// isFDv1FallbackRequested reports whether the response headers signal that the SDK should revert
+// to the FDv1 protocol.
+func isFDv1FallbackRequested(h http.Header) bool {
+	return h.Get(fdv1FallbackHeader) == "true"
+}
+
 type httpStatusError struct {
 	Message string
 	Code    int

internal/datasourcev2/polling_data_source.go

@@ -68,13 +68,14 @@ func (pp *PollingProcessor) Name() string {
 }
 
 //nolint:revive // DataInitializer method.
-func (pp *PollingProcessor) Fetch(ds subsystems.DataSelector, ctx context.Context) (*subsystems.Basis, error) {
+func (pp *PollingProcessor) Fetch(ds subsystems.DataSelector, ctx context.Context) (*subsystems.Basis, bool, error) {
 	changeSet, headers, err := pp.requester.Request(ctx, ds.Selector())
+	fallback := isFDv1FallbackRequested(headers)
 	if err != nil {
-		return nil, err
+		return nil, fallback, err
 	}
 	environmentID := internal.NewInitMetadataFromHeaders(headers).GetEnvironmentID()
-	return &subsystems.Basis{ChangeSet: *changeSet, Persist: true, EnvironmentID: environmentID}, nil
+	return &subsystems.Basis{ChangeSet: *changeSet, Persist: true, EnvironmentID: environmentID}, fallback, nil
 }
 
 //nolint:revive // DataSynchronizer method.
@@ -116,7 +117,7 @@ func (pp *PollingProcessor) Sync(ds subsystems.DataSelector) <-chan subsystems.D
 							Time:       time.Now(),
 						}
 
-						if hse.Header.Get("X-LD-FD-Fallback") == "true" {
+						if isFDv1FallbackRequested(hse.Header) {
 							resultChan <- subsystems.DataSynchronizerResult{
 								State:         interfaces.DataSourceStateOff,
 								Error:         errorInfo,

internal/datasourcev2/polling_data_source_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/launchdarkly/go-server-sdk/v7/testhelpers/ldservicesv2"
 	"github.com/launchdarkly/go-test-helpers/v3/httphelpers"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 var alwaysTrueFlag = ldbuilders.NewFlagBuilder("always-true-flag").SingleVariation(ldvalue.Bool(true)).Build()
@@ -40,7 +41,7 @@ func TestPollingProcessorInitializerCanMakeSuccessfulRequest(t *testing.T) {
 		)
 		defer processor.Close()
 
-		basis, err := processor.Fetch(ds, context.Background())
+		basis, _, err := processor.Fetch(ds, context.Background())
 		assert.NoError(t, err)
 		assert.Len(t, basis.ChangeSet.Changes(), 1)
 		assert.Equal(t, basis.ChangeSet.IntentCode(), subsystems.IntentTransferFull)
@@ -67,7 +68,7 @@ func TestPollingProcessorInitializerAppendsFilterParameter(t *testing.T) {
 			},
 		)
 		defer processor.Close()
-		_, err := processor.Fetch(ds, context.Background())
+		_, _, err := processor.Fetch(ds, context.Background())
 		assert.NoError(t, err)
 
 		r := <-requestsCh
@@ -90,7 +91,7 @@ func TestPollingProcessorInitializerAppendsBasisParameter(t *testing.T) {
 		defer processor.Close()
 
 		ds := mocks.NewMockDataSelector(subsystems.NewSelector("test-state", 1))
-		_, err := processor.Fetch(ds, context.Background())
+		_, _, err := processor.Fetch(ds, context.Background())
 		assert.NoError(t, err)
 
 		r := <-requestsCh
@@ -115,7 +116,7 @@ func TestPollingProcessorSynchronizerAppendsFilterParameter(t *testing.T) {
 		)
 		defer processor.Close()
 
-		_, err := processor.Fetch(ds, context.Background())
+		_, _, err := processor.Fetch(ds, context.Background())
 		assert.NoError(t, err)
 
 		r := <-requestsCh
@@ -261,6 +262,83 @@ func TestPollingProcessorSynchronizerHandlesFallbackToFDv2(t *testing.T) {
 	})
 }
 
+func TestPollingProcessorInitializerHandlesFallbackOnSuccessfulResponse(t *testing.T) {
+	ds := mocks.NewMockDataSelector(subsystems.NoSelector())
+	data := ldservicesv2.NewServerSDKData().Flags(alwaysTrueFlag).ToInitializerPayload(subsystems.NewSelector("test-state", 1))
+
+	// Wrap the valid 200 handler to inject X-LD-FD-Fallback alongside a well-formed payload.
+	underlying := ldservices.ServerSidePollingV2ServiceHandler(data)
+	fallbackOn200 := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("X-LD-FD-Fallback", "true")
+		underlying.ServeHTTP(w, r)
+	})
+
+	httphelpers.WithServer(fallbackOn200, func(ts *httptest.Server) {
+		processor := NewPollingProcessor(
+			sharedtest.BasicClientContext(),
+			datasource.PollingConfig{
+				BaseURI:      ts.URL,
+				PollInterval: time.Minute * 30,
+			},
+		)
+		defer processor.Close()
+
+		basis, fallback, err := processor.Fetch(ds, context.Background())
+		assert.True(t, fallback)
+		assert.NoError(t, err)
+		// Even when the server signals fallback, a valid payload in the same response
+		// should still be surfaced so the caller can apply it before switching protocols.
+		require.NotNil(t, basis)
+		assert.Len(t, basis.ChangeSet.Changes(), 1)
+		assert.Equal(t, "test-state", basis.ChangeSet.Selector().State())
+	})
+}
+
+func TestPollingProcessorInitializerHandlesFallbackOnErrorResponse(t *testing.T) {
+	ds := mocks.NewMockDataSelector(subsystems.NoSelector())
+
+	fallbackHeader := http.Header{
+		"X-LD-FD-Fallback": []string{"true"},
+	}
+	handler, _ := httphelpers.RecordingHandler(httphelpers.HandlerWithResponse(500, fallbackHeader, nil))
+	httphelpers.WithServer(handler, func(ts *httptest.Server) {
+		processor := NewPollingProcessor(
+			sharedtest.BasicClientContext(),
+			datasource.PollingConfig{
+				BaseURI:      ts.URL,
+				PollInterval: time.Minute * 30,
+			},
+		)
+		defer processor.Close()
+
+		basis, fallback, err := processor.Fetch(ds, context.Background())
+		assert.Nil(t, basis)
+		assert.True(t, fallback)
+		assert.Error(t, err, "underlying HTTP error should still be surfaced alongside the fallback signal")
+	})
+}
+
+func TestPollingProcessorInitializerErrorWithoutFallbackHeaderReturnsRegularError(t *testing.T) {
+	ds := mocks.NewMockDataSelector(subsystems.NoSelector())
+
+	handler, _ := httphelpers.RecordingHandler(httphelpers.HandlerWithStatus(500))
+	httphelpers.WithServer(handler, func(ts *httptest.Server) {
+		processor := NewPollingProcessor(
+			sharedtest.BasicClientContext(),
+			datasource.PollingConfig{
+				BaseURI:      ts.URL,
+				PollInterval: time.Minute * 30,
+			},
+		)
+		defer processor.Close()
+
+		basis, fallback, err := processor.Fetch(ds, context.Background())
+		assert.Nil(t, basis)
+		assert.False(t, fallback)
+		assert.Error(t, err)
+	})
+}
+
 func TestPollingProcessorSynchronizerHandlesRecoverableErrors(t *testing.T) {
 	for _, statusCode := range []int{400, 408, 429, 500, 503} {
 		t.Run(fmt.Sprintf("handles recoverable error %d", statusCode), func(t *testing.T) {

internal/datasourcev2/streaming_data_source.go

@@ -106,8 +106,8 @@ func (sp *StreamProcessor) Name() string {
 }
 
 //nolint:revive // DataInitializer method.
-func (sp *StreamProcessor) Fetch(ds subsystems.DataSelector, _ context.Context) (*subsystems.Basis, error) {
-	return nil, errors.New("StreamProcessor does not implement Fetch capability")
+func (sp *StreamProcessor) Fetch(ds subsystems.DataSelector, _ context.Context) (*subsystems.Basis, bool, error) {
+	return nil, false, errors.New("StreamProcessor does not implement Fetch capability")
 }
 
 //nolint:revive // DataSynchronizer method.
@@ -353,7 +353,7 @@ func (sp *StreamProcessor) subscribe(ds subsystems.DataSelector, resultChan chan
 				Time:       time.Now(),
 			}
 
-			if se.Header.Get("X-LD-FD-Fallback") == "true" {
+			if isFDv1FallbackRequested(se.Header) {
 				resultChan <- subsystems.DataSynchronizerResult{
 					State:         interfaces.DataSourceStateOff,
 					Error:         errorInfo,

internal/datasourcev2/streaming_data_source_test.go

@@ -60,8 +60,9 @@ func TestStreamingDoesNotWorkAsInitializer(t *testing.T) {
 	)
 
 	defer sp.Close()
-	basis, err := sp.Fetch(ds, context.Background())
+	basis, fallback, err := sp.Fetch(ds, context.Background())
 	assert.Nil(t, basis)
+	assert.False(t, fallback)
 	assert.NotNil(t, err)
 }
 

internal/datasystem/fdv2_datasystem.go

@@ -205,7 +205,17 @@ func (f *FDv2) launchTask(task func()) {
 func (f *FDv2) run(ctx context.Context, closeWhenReady chan struct{}) {
 	f.UpdateStatus(interfaces.DataSourceStateInitializing, interfaces.DataSourceErrorInfo{})
 
-	f.runInitializers(ctx, closeWhenReady)
+	if f.runInitializers(ctx, closeWhenReady) {
+		if f.fdv1FallbackBuilder != nil {
+			f.loggers.Warn("Reverting to FDv1 protocol")
+			f.synchronizerBuilders = []func() (subsystems.DataSynchronizer, error){f.fdv1FallbackBuilder}
+			f.currentSyncIndex = 0
+		} else {
+			f.loggers.Warn("Initializer requested FDv1 fallback but none configured")
+			f.synchronizerBuilders = nil
+			f.UpdateStatus(interfaces.DataSourceStateOff, f.getStatus().LastError)
+		}
+	}
 
 	if f.configuredWithDataSources && f.dataStoreStatusProvider.IsStatusMonitoringEnabled() {
 		f.launchTask(func() {
@@ -235,12 +245,33 @@ func (f *FDv2) runPersistentStoreOutageRecovery(ctx context.Context, statuses <-
 	}
 }
 
-func (f *FDv2) runInitializers(ctx context.Context, closeWhenReady chan struct{}) {
+// runInitializers runs each configured initializer in order until one succeeds, the context is
+// cancelled, or an initializer signals a revert to FDv1. Returns true if an initializer signalled
+// a revert to FDv1. If fallback is signalled alongside a valid Basis, that Basis is applied before
+// returning so evaluations can serve the server-provided data while the FDv1 synchronizer spins up.
+func (f *FDv2) runInitializers(ctx context.Context, closeWhenReady chan struct{}) (fallbackToFDv1 bool) {
 	for _, initializer := range f.initializers {
 		f.loggers.Infof("Attempting to initialize via %s", initializer.Name())
-		basis, err := initializer.Fetch(f.store, ctx)
+		basis, fallback, err := initializer.Fetch(f.store, ctx)
 		if errors.Is(err, context.Canceled) {
-			return
+			return false
+		}
+		if fallback {
+			if err != nil {
+				f.loggers.Warnf("Initializer %s requested fallback to FDv1 protocol: %v", initializer.Name(), err)
+			} else {
+				f.loggers.Warnf("Initializer %s requested fallback to FDv1 protocol", initializer.Name())
+			}
+			if basis != nil {
+				f.environmentIDProvider.SetEnvironmentID(basis.EnvironmentID)
+				f.store.Apply(basis.ChangeSet, basis.Persist)
+				if basis.ChangeSet.Selector().IsDefined() {
+					f.readyOnce.Do(func() {
+						close(closeWhenReady)
+					})
+				}
+			}
+			return true
 		}
 		if err != nil {
 			f.loggers.Warnf("Initializer %s failed: %v", initializer.Name(), err)
@@ -253,9 +284,10 @@ func (f *FDv2) runInitializers(ctx context.Context, closeWhenReady chan struct{}
 			f.readyOnce.Do(func() {
 				close(closeWhenReady)
 			})
-			return
+			return false
 		}
 	}
+	return false
 }
 
 func (f *FDv2) runSynchronizers(ctx context.Context, closeWhenReady chan struct{}) {

ldclient_end_to_end_fdv2_test.go

@@ -116,6 +116,97 @@ func TestFDV2CanFallBackToV1(t *testing.T) {
 	})
 }
 
+// When an initializer requests FDv1 fallback but no FDv1 fallback is configured, the data source
+// status must transition to Off rather than staying stuck at Initializing. This mirrors the
+// synchronizer-triggered path when fdv1FallbackBuilder is nil.
+func TestFDV2InitializerFallbackWithoutFDv1FallbackTransitionsToOff(t *testing.T) {
+	header := http.Header{
+		"X-LD-FD-Fallback": []string{"true"},
+	}
+	handler, requestsCh := httphelpers.RecordingHandler(httphelpers.HandlerWithResponse(500, header, nil))
+
+	httphelpers.WithServer(handler, func(server *httptest.Server) {
+		logCapture := ldlogtest.NewMockLog()
+
+		// Custom data system: a polling initializer, no synchronizers, no FDv1 fallback.
+		config := Config{
+			Events:  ldcomponents.NoEvents(),
+			Logging: ldcomponents.Logging().Loggers(logCapture.Loggers),
+			DataSystem: ldcomponents.DataSystem().Custom().Initializers(
+				ldcomponents.PollingDataSourceV2().BaseURI(server.URL).AsInitializer(),
+			),
+		}
+
+		client, err := MakeCustomClient(testSdkKey, config, time.Second*5)
+		require.Error(t, err)
+		require.NotNil(t, client)
+		defer client.Close()
+
+		<-requestsCh
+
+		// With no FDv1 fallback configured, an initializer-triggered fallback must transition the
+		// status to Off — if it stays at Initializing, MakeCustomClient treats it as an init
+		// failure and we see initializationFailedErrorMessage here. Either way the status field
+		// should end up Off, so assert that directly.
+		assert.Equal(t,
+			interfaces.DataSourceStateOff,
+			client.GetDataSourceStatusProvider().GetStatus().State,
+			"status should transition to Off when initializer fallback requested but no FDv1 fallback configured")
+		assert.Equal(t, initializationFailedErrorMessage, err.Error())
+
+		assert.Contains(t, logCapture.GetOutput(ldlog.Warn),
+			"Initializer requested FDv1 fallback but none configured")
+	})
+}
+
+// When the polling initializer receives x-ld-fd-fallback from the server, the SDK should skip any
+// remaining FDv2 synchronizers and switch to the FDv1 polling synchronizer directly — without ever
+// attempting the FDv2 streaming synchronizer.
+func TestFDV2CanFallBackToV1FromInitializer(t *testing.T) {
+	dataV1 := ldservices.NewServerSDKData().Flags(alwaysFalseFlag)
+
+	header := http.Header{
+		"X-LD-FD-Fallback": []string{"true"},
+	}
+
+	// FDv2 polling initializer: returns 500 + fallback header. Must trigger revert to FDv1 before
+	// the FDv2 streaming synchronizer is ever dialed.
+	pollV2InitRecordingHandler, pollV2InitReqCh := httphelpers.RecordingHandler(httphelpers.HandlerWithResponse(500, header, nil))
+	// FDv1 polling synchronizer: returns valid FDv1 data.
+	pollV1SyncRecordingHandler, pollV1SyncReqCh := httphelpers.RecordingHandler(ldservices.ServerSidePollingServiceHandler(dataV1))
+	// FDv2 streaming synchronizer: should never be hit. If it is, the test will surface it via
+	// streamV2SyncReqCh.
+	streamHandler, streamV2SyncReqCh := httphelpers.RecordingHandler(httphelpers.HandlerWithResponse(500, header, nil))
+
+	handler := httphelpers.SequentialHandler(pollV2InitRecordingHandler, pollV1SyncRecordingHandler, streamHandler)
+
+	httphelpers.WithServer(handler, func(server *httptest.Server) {
+		logCapture := ldlogtest.NewMockLog()
+
+		config := Config{
+			Events:     ldcomponents.NoEvents(),
+			Logging:    ldcomponents.Logging().Loggers(logCapture.Loggers),
+			DataSystem: ldcomponents.DataSystem().WithRelayProxyEndpoints(server.URL).Default(),
+		}
+
+		client, err := MakeCustomClient(testSdkKey, config, time.Second*5)
+
+		<-pollV2InitReqCh
+		<-pollV1SyncReqCh
+
+		require.NoError(t, err)
+		defer client.Close()
+
+		reached := client.GetDataSourceStatusProvider().WaitFor(interfaces.DataSourceStateValid, time.Second*5)
+		require.True(t, reached, "timed out waiting for data source to reach VALID state")
+
+		assertNoMoreRequests(t, streamV2SyncReqCh)
+
+		value, _ := client.BoolVariation(alwaysFalseFlag.Key, testUser, true)
+		assert.False(t, value)
+	})
+}
+
 func TestFDV2StreamingSynchronizer(t *testing.T) {
 	data := ldservicesv2.NewServerSDKData().Flags(alwaysTrueFlag)
 

ldfiledatav2/file_data_source_impl.go

@@ -135,7 +135,7 @@ func (fs *fileDataSource) Sync(ds subsystems.DataSelector) <-chan subsystems.Dat
 	return resultChan
 }
 
-func (fs *fileDataSource) Fetch(ds subsystems.DataSelector, ctx context.Context) (*subsystems.Basis, error) {
+func (fs *fileDataSource) Fetch(ds subsystems.DataSelector, ctx context.Context) (*subsystems.Basis, bool, error) {
 	changeSetChan := fs.changeSetBroadcaster.AddListener()
 	statusChan := fs.statusBroadcaster.AddListener()
 
@@ -178,7 +178,7 @@ func (fs *fileDataSource) Fetch(ds subsystems.DataSelector, ctx context.Context)
 
 	wg.Wait()
 
-	return basis, err
+	return basis, false, err
 }
 
 // Reload tells the data source to immediately attempt to reread all of the configured source files