fix: Make FDv1 fallback TTL-based and always recover to FDv2

internal/datasourcev2/fdv1_polling_data_source.go

@@ -3,6 +3,7 @@ package datasourcev2
 import (
 	"context"
 	"net/http"
+	"time"
 
 	"github.com/launchdarkly/go-sdk-common/v3/ldlog"
 	"github.com/launchdarkly/go-server-sdk/v7/internal/datasource"
@@ -17,10 +18,10 @@ type fdv1ToFDv2Requester struct {
 func (r *fdv1ToFDv2Requester) Request(
 	ctx context.Context,
 	selector subsystems.Selector,
-) (*subsystems.ChangeSet, http.Header, error) {
+) (changeSet *subsystems.ChangeSet, headers http.Header, fdv1FallbackTTL *time.Duration, err error) {
 	data, cached, headers, err := r.requester.Request()
 	if err != nil {
-		return nil, headers, err
+		return nil, headers, nil, err
 	}
 
 	code := subsystems.IntentTransferFull
@@ -47,7 +48,7 @@ func (r *fdv1ToFDv2Requester) Request(
 		subsystems.NewSelector("state", 1),
 		data,
 	)
-	return changeset, headers, err
+	return changeset, headers, nil, err
 }
 
 func (r *fdv1ToFDv2Requester) BaseURI() string {

internal/datasourcev2/helpers.go

@@ -6,10 +6,15 @@ package datasourcev2
 // v2 implementation, we should clean this up.
 
 import (
+	"encoding/json"
 	"fmt"
+	"math"
 	"net/http"
+	"strconv"
+	"time"
 
 	"github.com/launchdarkly/go-sdk-common/v3/ldlog"
+	"github.com/launchdarkly/go-server-sdk/v7/subsystems"
 )
 
 // fdv1FallbackHeader is the response header LaunchDarkly uses to instruct an SDK to abandon the
@@ -22,6 +27,48 @@ func isFDv1FallbackRequested(h http.Header) bool {
 	return h.Get(fdv1FallbackHeader) == "true"
 }
 
+// fdv1FallbackTTLHeader is the response header that says how many seconds the SDK should stay
+// on FDv1 before trying FDv2 again.
+const fdv1FallbackTTLHeader = "X-LD-FD-Fallback-TTL"
+
+// fdv1FallbackFromHeaders returns the FDv1 fallback instruction carried by the response headers.
+func fdv1FallbackFromHeaders(h http.Header) *time.Duration {
+	if !isFDv1FallbackRequested(h) {
+		return nil
+	}
+	seconds, err := strconv.ParseInt(h.Get(fdv1FallbackTTLHeader), 10, 64)
+	if err != nil {
+		return new(time.Duration)
+	}
+	return ttlPointer(fdv1FallbackTTLFromSeconds(seconds))
+}
+
+// fdv1FallbackFromGoodbye returns the FDv1 fallback instruction carried by a goodbye event. A
+// goodbye event is an instruction to fall back only if it has a protocolFallbackTTL.
+func fdv1FallbackFromGoodbye(goodbye subsystems.Goodbye) *time.Duration {
+	if len(goodbye.ProtocolFallbackTTL) == 0 || string(goodbye.ProtocolFallbackTTL) == "null" {
+		return nil
+	}
+	var seconds int64
+	if err := json.Unmarshal(goodbye.ProtocolFallbackTTL, &seconds); err != nil {
+		return new(time.Duration)
+	}
+	return ttlPointer(fdv1FallbackTTLFromSeconds(seconds))
+}
+
+func ttlPointer(ttl time.Duration) *time.Duration {
+	return &ttl
+}
+
+// fdv1FallbackTTLFromSeconds converts a TTL in seconds to a duration. A value that is too large to
+// convert gives zero, which means that the TTL is not usable.
+func fdv1FallbackTTLFromSeconds(seconds int64) time.Duration {
+	if seconds > math.MaxInt64/int64(time.Second) {
+		return 0
+	}
+	return time.Duration(seconds) * time.Second
+}
+
 type httpStatusError struct {
 	Message string
 	Code    int

internal/datasourcev2/helpers_test.go

@@ -1,10 +1,16 @@
 package datasourcev2
 
 import (
+	"encoding/json"
+	"net/http"
 	"strconv"
 	"testing"
+	"time"
+
+	"github.com/launchdarkly/go-server-sdk/v7/subsystems"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestHTTPStatusError(t *testing.T) {
@@ -51,3 +57,57 @@ func testWithFilters(t *testing.T, test func(t *testing.T, filterTest filterTest
 		})
 	}
 }
+
+func TestFDv1FallbackFromHeaders(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		fallback string
+		ttl      string
+		want     *time.Duration
+	}{
+		{name: "no fallback header", ttl: "120", want: nil},
+		{name: "no TTL header", fallback: "true", want: new(time.Duration)},
+		{name: "valid TTL", fallback: "true", ttl: "120", want: ttlPointer(120 * time.Second)},
+		{name: "malformed TTL", fallback: "true", ttl: "soon", want: new(time.Duration)},
+		{name: "fractional TTL", fallback: "true", ttl: "1.5", want: new(time.Duration)},
+		{name: "negative TTL", fallback: "true", ttl: "-5", want: ttlPointer(-5 * time.Second)},
+		{name: "huge TTL", fallback: "true", ttl: "9223372036854775807", want: new(time.Duration)},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			h := http.Header{}
+			if tc.fallback != "" {
+				h.Set(fdv1FallbackHeader, tc.fallback)
+			}
+			if tc.ttl != "" {
+				h.Set(fdv1FallbackTTLHeader, tc.ttl)
+			}
+			assert.Equal(t, tc.want, fdv1FallbackFromHeaders(h))
+		})
+	}
+}
+
+func TestFDv1FallbackFromGoodbye(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		ttl  string
+		want *time.Duration
+	}{
+		{name: "no TTL", want: nil},
+		{name: "null TTL", ttl: "null", want: nil},
+		{name: "valid TTL", ttl: "120", want: ttlPointer(120 * time.Second)},
+		{name: "zero TTL", ttl: "0", want: new(time.Duration)},
+		{name: "malformed TTL", ttl: `"soon"`, want: new(time.Duration)},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			goodbye := subsystems.Goodbye{Reason: "test", ProtocolFallbackTTL: json.RawMessage(tc.ttl)}
+			assert.Equal(t, tc.want, fdv1FallbackFromGoodbye(goodbye))
+		})
+	}
+}
+
+func TestGoodbyeUnmarshalsProtocolFallbackTTL(t *testing.T) {
+	var goodbye subsystems.Goodbye
+	require.NoError(t, json.Unmarshal([]byte(`{"reason":"r","protocolFallbackTTL":120}`), &goodbye))
+
+	assert.Equal(t, ttlPointer(120*time.Second), fdv1FallbackFromGoodbye(goodbye))
+}

internal/datasourcev2/polling_data_source.go

@@ -21,7 +21,14 @@ const (
 // PollingRequester allows PollingProcessor to delegate fetching data to another component.
 // This is useful for testing the PollingProcessor without needing to set up a test HTTP server.
 type PollingRequester interface {
-	Request(context.Context, subsystems.Selector) (*subsystems.ChangeSet, http.Header, error)
+	// Request fetches a change set. If LaunchDarkly instructed the SDK to fall back to FDv1,
+	// fdv1FallbackTTL is not nil; zero means that no usable TTL was supplied.
+	Request(ctx context.Context, selector subsystems.Selector) (
+		changeSet *subsystems.ChangeSet,
+		headers http.Header,
+		fdv1FallbackTTL *time.Duration,
+		err error,
+	)
 	BaseURI() string
 	FilterKey() string
 }
@@ -68,9 +75,11 @@ func (pp *PollingProcessor) Name() string {
 }
 
 //nolint:revive // DataInitializer method.
-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)
+func (pp *PollingProcessor) Fetch(
+	ds subsystems.DataSelector,
+	ctx context.Context,
+) (*subsystems.Basis, *time.Duration, error) {
+	changeSet, headers, fallback, err := pp.requester.Request(ctx, ds.Selector())
 	if err != nil {
 		return nil, fallback, err
 	}
@@ -110,8 +119,8 @@ func (pp *PollingProcessor) Sync(ds subsystems.DataSelector) <-chan subsystems.D
 				result, err := pp.poll(ctx, ds)
 
 				// When the server requested FDv1 fallback, dispatch the result as-is -- poll has
-				// already populated State (Valid on success, Off on error) and FallbackToFDv1=true.
-				if result.FallbackToFDv1 {
+				// already populated State (Valid on success, Off on error) and FDv1FallbackTTL.
+				if result.FDv1FallbackTTL != nil {
 					resultChan <- result
 					return
 				}
@@ -150,9 +159,9 @@ func (pp *PollingProcessor) Sync(ds subsystems.DataSelector) <-chan subsystems.D
 }
 
 // poll performs a single polling request and builds a DataSynchronizerResult describing the
-// outcome. The result's FallbackToFDv1 flag is always populated from the x-ld-fd-fallback response
-// header, whether or not the request succeeded -- a 500 or a malformed-JSON body can still carry
-// the fallback signal.
+// outcome. The result's FDv1FallbackTTL is always populated from the
+// fallback instruction in the response, whether or not the request succeeded -- a 500 or a
+// malformed-JSON body can still carry the fallback signal.
 //
 // On success: result.State = Valid, result.ChangeSet populated, err = nil.
 // On error: result.State = Off (the safer default), result.Error populated with Kind/Message/
@@ -164,10 +173,10 @@ func (pp *PollingProcessor) Sync(ds subsystems.DataSelector) <-chan subsystems.D
 func (pp *PollingProcessor) poll(
 	ctx context.Context, ds subsystems.DataSelector,
 ) (subsystems.DataSynchronizerResult, error) {
-	changeSet, headers, err := pp.requester.Request(ctx, ds.Selector())
+	changeSet, headers, fallback, err := pp.requester.Request(ctx, ds.Selector())
 	result := subsystems.DataSynchronizerResult{
-		EnvironmentID:  internal.NewInitMetadataFromHeaders(headers).GetEnvironmentID(),
-		FallbackToFDv1: isFDv1FallbackRequested(headers),
+		EnvironmentID:   internal.NewInitMetadataFromHeaders(headers).GetEnvironmentID(),
+		FDv1FallbackTTL: fallback,
 	}
 
 	if err == nil {
@@ -189,8 +198,11 @@ func (pp *PollingProcessor) poll(
 			Message: err.Error(),
 			Time:    time.Now(),
 		}
-		if _, ok := err.(malformedJSONError); ok {
+		switch err.(type) {
+		case malformedJSONError:
 			result.Error.Kind = interfaces.DataSourceErrorKindInvalidData
+		case goodbyeError:
+			result.Error.Kind = interfaces.DataSourceErrorKindUnknown
 		}
 	}
 	return result, err

internal/datasourcev2/polling_data_source_test.go

@@ -2,6 +2,7 @@ package datasourcev2
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"net/http"
 	"net/http/httptest"
@@ -259,13 +260,13 @@ func TestPollingProcessorSynchronizerHandlesFallbackOnSuccessfulResponse(t *test
 
 		resultChan := processor.Sync(ds)
 
-		// A single Valid result carries both the payload and the FallbackToFDv1 signal -- the
+		// A single Valid result carries both the payload and the FDv1FallbackTTL signal -- the
 		// consumer applies the ChangeSet first, then switches to the FDv1 synchronizer.
 		result := <-resultChan
 		assert.Equal(t, interfaces.DataSourceStateValid, result.State)
 		require.NotNil(t, result.ChangeSet)
 		assert.Len(t, result.ChangeSet.Changes(), 1)
-		assert.True(t, result.FallbackToFDv1)
+		require.NotNil(t, result.FDv1FallbackTTL)
 	})
 }
 
@@ -293,15 +294,16 @@ func TestPollingProcessorSynchronizerHandlesFallbackOnMalformedBody(t *testing.T
 
 		assert.Equal(t, interfaces.DataSourceStateOff, result.State)
 		assert.Equal(t, interfaces.DataSourceErrorKindInvalidData, result.Error.Kind)
-		assert.True(t, result.FallbackToFDv1)
+		require.NotNil(t, result.FDv1FallbackTTL)
 	})
 }
 
 func TestPollingProcessorSynchronizerHandlesFallbackToFDv2(t *testing.T) {
 	ds := mocks.NewMockDataSelector(subsystems.NoSelector())
 
 	header := http.Header{
-		"X-LD-FD-Fallback": []string{"true"},
+		"X-LD-FD-Fallback":     []string{"true"},
+		"X-LD-FD-Fallback-TTL": []string{"120"},
 	}
 	handler, _ := httphelpers.RecordingHandler(httphelpers.HandlerWithResponse(500, header, nil))
 	httphelpers.WithServer(handler, func(ts *httptest.Server) {
@@ -319,7 +321,41 @@ func TestPollingProcessorSynchronizerHandlesFallbackToFDv2(t *testing.T) {
 
 		assert.Equal(t, result.State, interfaces.DataSourceStateOff)
 		assert.Equal(t, result.Error.Kind, interfaces.DataSourceErrorKindErrorResponse)
-		assert.True(t, result.FallbackToFDv1)
+		require.NotNil(t, result.FDv1FallbackTTL)
+		assert.Equal(t, 120*time.Second, *result.FDv1FallbackTTL)
+	})
+}
+
+func TestPollingProcessorSynchronizerFallsBackOnGoodbyeWithFallbackTTL(t *testing.T) {
+	ds := mocks.NewMockDataSelector(subsystems.NoSelector())
+
+	payload := subsystems.PollingPayload{Events: []subsystems.RawEvent{
+		{
+			Name: subsystems.EventServerIntent,
+			Data: json.RawMessage(`{"payloads":[{"id":"id","target":1,"intentCode":"xfer-full","reason":"r"}]}`),
+		},
+		{
+			Name: subsystems.EventGoodbye,
+			Data: json.RawMessage(`{"reason":"fall back","protocolFallbackTTL":120}`),
+		},
+	}}
+
+	httphelpers.WithServer(ldservices.ServerSidePollingV2ServiceHandler(payload), func(ts *httptest.Server) {
+		processor := NewPollingProcessor(
+			sharedtest.BasicClientContext(),
+			datasource.PollingConfig{
+				BaseURI:      ts.URL,
+				PollInterval: time.Minute * 30,
+			},
+		)
+		defer processor.Close()
+
+		result := <-processor.Sync(ds)
+
+		assert.Equal(t, interfaces.DataSourceStateOff, result.State)
+		assert.Equal(t, interfaces.DataSourceErrorKindUnknown, result.Error.Kind)
+		require.NotNil(t, result.FDv1FallbackTTL)
+		assert.Equal(t, 120*time.Second, *result.FDv1FallbackTTL)
 	})
 }
 
@@ -345,7 +381,7 @@ func TestPollingProcessorInitializerHandlesFallbackOnSuccessfulResponse(t *testi
 		defer processor.Close()
 
 		basis, fallback, err := processor.Fetch(ds, context.Background())
-		assert.True(t, fallback)
+		assert.NotNil(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.
@@ -374,7 +410,7 @@ func TestPollingProcessorInitializerHandlesFallbackOnErrorResponse(t *testing.T)
 
 		basis, fallback, err := processor.Fetch(ds, context.Background())
 		assert.Nil(t, basis)
-		assert.True(t, fallback)
+		assert.NotNil(t, fallback)
 		assert.Error(t, err, "underlying HTTP error should still be surfaced alongside the fallback signal")
 	})
 }
@@ -395,7 +431,7 @@ func TestPollingProcessorInitializerErrorWithoutFallbackHeaderReturnsRegularErro
 
 		basis, fallback, err := processor.Fetch(ds, context.Background())
 		assert.Nil(t, basis)
-		assert.False(t, fallback)
+		assert.Nil(t, fallback)
 		assert.Error(t, err)
 	})
 }