feat: End an FDv1 fallback after a default TTL of one hour

internal/datasystem/fdv2_datasystem.go

@@ -102,6 +102,9 @@ type FDv2 struct {
 	fallbackCond func(status interfaces.DataSourceStatus) bool
 	recoveryCond func(status interfaces.DataSourceStatus) bool
 
+	// The default duration of a directed fallback to FDv1, before jitter.
+	defaultFDv1FallbackTTL time.Duration
+
 	// The following are non-nil only when an override source is configured. The layer holds
 	// override entries. The overlay serves them in preference to the store's data at the
 	// store read boundary. The source populates the layer at runtime. The override system is
@@ -138,6 +141,7 @@ func NewFDv2(disabled bool, cfgBuilder subsystems.ComponentConfigurer[subsystems
 		dataSourceStatusProvider: &dataStatusProvider{},
 		environmentIDProvider:    &environmentIDProvider{},
 		sdkKeyChanged:            make(chan struct{}, 1),
+		defaultFDv1FallbackTTL:   defaultFDv1FallbackTTL,
 	}
 
 	// Unfortunate circular reference.
@@ -245,16 +249,16 @@ func (f *FDv2) run(ctx context.Context, closeWhenReady chan struct{}) {
 	f.UpdateStatus(interfaces.DataSourceStateInitializing, interfaces.DataSourceErrorInfo{})
 
 	if fallback, errorInfo := f.runInitializers(ctx, closeWhenReady); fallback {
+		ttl := jittered(f.defaultFDv1FallbackTTL)
 		if f.synchronizers.hasFDv1Fallback() {
-			f.loggers.Warn("Falling back to FDv1 protocol")
-			// Exclude every FDv2 synchronizer and make the FDv1 fallback synchronizer available.
-			f.synchronizers.fallBackToFDv1()
+			f.loggers.Warnf("Falling back to FDv1 protocol, retrying the configured synchronizers in %s", ttl)
 		} else {
-			f.loggers.Warn("Initializer requested FDv1 fallback but none configured")
-			// Exclude every synchronizer so that none starts.
-			f.synchronizers.excludeAll()
+			f.loggers.Warnf(
+				"Initializer requested FDv1 fallback but none configured, retrying the configured synchronizers in %s", ttl)
 			f.UpdateStatus(interfaces.DataSourceStateInterrupted, errorInfo)
 		}
+		// Exclude every FDv2 synchronizer and make the FDv1 fallback synchronizer available, if there is one.
+		f.synchronizers.fallBackToFDv1(ttl)
 	}
 
 	if f.configuredWithDataSources && f.dataStoreStatusProvider.IsStatusMonitoringEnabled() {
@@ -376,8 +380,9 @@ func (f *FDv2) completeInitialization(closeWhenReady chan struct{}) {
 }
 
 func (f *FDv2) runSynchronizers(ctx context.Context, closeWhenReady chan struct{}) {
-	// If no synchronizers are available, close ready channel and return
-	if f.synchronizers.eligibleCount() == 0 {
+	// If no synchronizer can become available, close ready channel and return
+	_, pending := f.synchronizers.earliestUnpark()
+	if f.synchronizers.eligibleCount() == 0 && !pending {
 		f.readyOnce.Do(func() {
 			close(closeWhenReady)
 		})
@@ -392,7 +397,7 @@ func (f *FDv2) runSynchronizers(ctx context.Context, closeWhenReady chan struct{
 
 		f.synchronizers.next()
 		for {
-			// Nothing is eligible. Wait out the soonest unpark, then start the preferred slot.
+			// Nothing is eligible. Wait for the soonest unpark or fallback end, then start the preferred slot.
 			if f.synchronizers.eligibleCount() == 0 {
 				unpark, parked := f.synchronizers.earliestUnpark()
 				if !parked {
@@ -411,6 +416,9 @@ func (f *FDv2) runSynchronizers(ctx context.Context, closeWhenReady chan struct{
 					f.synchronizers.unparkAll()
 				case <-time.After(delay):
 				}
+				if f.synchronizers.maybeEndFDv1Fallback() {
+					f.loggers.Info("FDv1 fallback ended, returning to the configured synchronizers")
+				}
 				f.synchronizers.reset()
 				f.synchronizers.next()
 				continue
@@ -445,16 +453,19 @@ func (f *FDv2) runSynchronizers(ctx context.Context, closeWhenReady chan struct{
 			// Handle action based on conditions
 			switch action {
 			case syncFDv1:
+				ttl := jittered(f.defaultFDv1FallbackTTL)
 				if f.synchronizers.hasFDv1Fallback() {
-					f.loggers.Warn("Falling back to FDv1 protocol")
-					// Exclude every FDv2 synchronizer and make the FDv1 fallback synchronizer available.
-					f.synchronizers.fallBackToFDv1()
-					f.synchronizers.next()
-					continue
+					f.loggers.Warnf("Falling back to FDv1 protocol, retrying the configured synchronizers in %s", ttl)
+				} else {
+					f.loggers.Warnf(
+						"Synchronizer requested FDv1 fallback but none configured, retrying the configured synchronizers in %s",
+						ttl)
+					f.UpdateStatus(interfaces.DataSourceStateInterrupted, f.getStatus().LastError)
 				}
-				f.loggers.Warn("Synchronizer requested FDv1 fallback but none configured")
-				f.UpdateStatus(interfaces.DataSourceStateInterrupted, f.getStatus().LastError)
-				return
+				// Exclude every FDv2 synchronizer and make the FDv1 fallback synchronizer available, if there is one.
+				f.synchronizers.fallBackToFDv1(ttl)
+				f.synchronizers.next()
+				continue
 			case syncPark:
 				delay := f.synchronizers.parkCurrent()
 				f.loggers.Warnf("Synchronizer %s is backing off for %s after an unexpected error (%s)",
@@ -477,6 +488,10 @@ func (f *FDv2) runSynchronizers(ctx context.Context, closeWhenReady chan struct{
 				f.loggers.Info("Recovery condition met, returning to first synchronizer")
 				f.synchronizers.reset()
 				f.synchronizers.next()
+			case syncFDv1FallbackEnded:
+				f.loggers.Info("FDv1 fallback ended, returning to the configured synchronizers")
+				f.synchronizers.reset()
+				f.synchronizers.next()
 			case syncFallback:
 				// Fallback: move to the next available synchronizer
 				f.loggers.Info("Fallback condition met, trying next synchronizer")
@@ -502,6 +517,7 @@ const (
 	syncPark
 	syncFDv1
 	syncRestart
+	syncFDv1FallbackEnded
 )
 
 // conditionAction returns the action the configured conditions call for, or syncNone. Recovery
@@ -602,6 +618,10 @@ func (f *FDv2) consumeSynchronizerResults(
 				return syncPark, nil
 			}
 		case <-ticker.C:
+			if f.synchronizers.maybeEndFDv1Fallback() {
+				return syncFDv1FallbackEnded, nil
+			}
+
 			// If there's only one available synchronizer, don't check conditions
 			if f.synchronizers.eligibleCount() == 1 {
 				continue

internal/datasystem/fdv2_fallback_test.go

@@ -353,3 +353,49 @@ func TestFDv2StartupFallsBackPastSilentSynchronizer(t *testing.T) {
 	assert.True(t, healthy.wasStarted(30*time.Second),
 		"should fall back past the silent synchronizer to the third")
 }
+
+func TestFDv2ReturnsToFDv2AfterFallbackTTLWithoutFDv1Fallback(t *testing.T) {
+	primary := newScriptedSynchronizer("primary")
+	system := makeSDKKeyTestFDv2(t, primary)
+	system.defaultFDv1FallbackTTL = 200 * time.Millisecond
+	system.Start(make(chan struct{}))
+
+	// The primary synchronizer directs a fallback to FDv1, but no FDv1 fallback is configured.
+	primary.requireRun(t) <- subsystems.DataSynchronizerResult{
+		State:          interfaces.DataSourceStateInterrupted,
+		Error:          interfaces.DataSourceErrorInfo{Kind: interfaces.DataSourceErrorKindErrorResponse, StatusCode: 503},
+		FallbackToFDv1: true,
+	}
+
+	// The data system is interrupted and keeps the error of the directive.
+	require.Eventually(t, func() bool {
+		return system.DataSourceStatusProvider().GetStatus().LastError.StatusCode == 503
+	}, time.Second, time.Millisecond)
+	assert.Equal(t, interfaces.DataSourceStateInterrupted, system.DataSourceStatusProvider().GetStatus().State)
+
+	// The primary synchronizer starts again after the TTL.
+	primary.requireRun(t)
+}
+
+// The data system checks the TTL on its ten-second condition ticker, so this test takes about ten seconds.
+func TestFDv2ReturnsToFDv2AfterFallbackTTLWhileFDv1FallbackRuns(t *testing.T) {
+	primary := newScriptedSynchronizer("primary")
+	fdv1 := newScriptedSynchronizer("fdv1")
+	system := makeSDKKeyTestFDv2WithFDv1Fallback(t, fdv1, primary)
+	system.defaultFDv1FallbackTTL = 200 * time.Millisecond
+	system.Start(make(chan struct{}))
+
+	// The primary synchronizer directs a fallback to FDv1, and the FDv1 synchronizer gets data.
+	primary.requireRun(t) <- subsystems.DataSynchronizerResult{
+		State:          interfaces.DataSourceStateInterrupted,
+		FallbackToFDv1: true,
+	}
+	fdv1.requireRun(t) <- validResult
+
+	// The primary synchronizer starts again after the TTL, on the next condition check.
+	select {
+	case <-primary.runs:
+	case <-time.After(15 * time.Second):
+		require.FailNow(t, "timed out waiting for the primary synchronizer to start again")
+	}
+}

internal/datasystem/fdv2_sdk_key_test.go

@@ -225,19 +225,21 @@ func TestFDv2KeyChangeDoesNotInterruptRunningSynchronizer(t *testing.T) {
 }
 
 func TestFDv2KeyChangeDoesNotUndoFallbackDirectiveWithoutFDv1Fallback(t *testing.T) {
-	// No FDv1 fallback is configured, so a fallback directive stops every synchronizer for good.
+	// No FDv1 fallback is configured, so a fallback directive stops every synchronizer until the TTL elapses.
 	primary := newScriptedSynchronizer("primary")
 	system := makeSDKKeyTestFDv2(t, primary)
-	ready := make(chan struct{})
-	system.Start(ready)
+	system.Start(make(chan struct{}))
 	primary.requireRun(t) <- subsystems.DataSynchronizerResult{
 		State:          interfaces.DataSourceStateInterrupted,
 		FallbackToFDv1: true,
 	}
-	requireReady(t, ready)
+	require.Eventually(t, func() bool {
+		return system.DataSourceStatusProvider().GetStatus().State == interfaces.DataSourceStateInterrupted
+	}, time.Second, time.Millisecond)
 
 	// A key change does not start a synchronizer, and the data system stays interrupted.
 	system.SDKKeyChanged()
+	requireKeyChangeTaken(t, system)
 	primary.assertNoRun(t)
 	assert.Equal(t, interfaces.DataSourceStateInterrupted, system.DataSourceStatusProvider().GetStatus().State)
 }

internal/datasystem/synchronizer_list.go

@@ -18,6 +18,9 @@ const (
 	afterParkDelay   = 1 * time.Second
 )
 
+// The default duration of a directed fallback to FDv1, before jitter.
+const defaultFDv1FallbackTTL = 1 * time.Hour
+
 var errNoCurrentSynchronizer = errors.New("no current synchronizer")
 
 // synchronizerSlot holds one synchronizer builder and its state.
@@ -36,19 +39,23 @@ type synchronizerSlot struct {
 	parkAttempts int
 }
 
+// jittered returns d minus a random jitter of up to half of d.
+func jittered(d time.Duration) time.Duration {
+	if half := int64(d / 2); half > 0 {
+		return d - time.Duration(rand.Int63n(half)) //nolint:gosec // weak RNG is acceptable for jitter
+	}
+	return d
+}
+
 // park parks the slot and returns the delay until its unpark time. Each successive park
 // doubles the delay, up to parkMaxDelay.
-func (s *synchronizerSlot) park(rng *rand.Rand) time.Duration {
+func (s *synchronizerSlot) park() time.Duration {
 	s.parkAttempts++
 	t := time.Duration(math.Min(
 		float64(parkInitialDelay)*math.Pow(2, float64(s.parkAttempts-1)),
 		float64(parkMaxDelay),
 	))
-	var jitter time.Duration
-	if halfT := int64(t / 2); halfT > 0 {
-		jitter = time.Duration(rng.Int63n(halfT))
-	}
-	delay := t - jitter
+	delay := jittered(t)
 	s.parkedUntil = time.Now().Add(delay)
 	return delay
 }
@@ -64,8 +71,8 @@ type synchronizerList struct {
 	// Index of the current slot, or -1 if there is no current slot.
 	current int
 
-	// Source of jitter for park delays.
-	rng *rand.Rand
+	// The end of the directed fallback to FDv1. Zero means no fallback is in effect.
+	fdv1FallbackUntil time.Time
 }
 
 func newSynchronizerList(builders []synchronizerBuilder, fdv1Fallback synchronizerBuilder) *synchronizerList {
@@ -80,8 +87,6 @@ func newSynchronizerList(builders []synchronizerBuilder, fdv1Fallback synchroniz
 		slots:   slots,
 		hasFDv1: fdv1Fallback != nil,
 		current: -1,
-		//nolint:gosec // not a cryptographic use-case, weak RNG is acceptable for jitter
-		rng: rand.New(rand.NewSource(time.Now().UnixNano())),
 	}
 }
 
@@ -134,7 +139,7 @@ func (l *synchronizerList) parkCurrent() time.Duration {
 	if l.current < 0 {
 		return 0
 	}
-	return l.slots[l.current].park(l.rng)
+	return l.slots[l.current].park()
 }
 
 // unparkAll unparks every slot, including synchronizers that are currently excluded.
@@ -146,7 +151,7 @@ func (l *synchronizerList) unparkAll() {
 
 // failoverDelay returns the delay to wait before starting another synchronizer after a park.
 func (l *synchronizerList) failoverDelay() time.Duration {
-	return afterParkDelay - time.Duration(l.rng.Int63n(int64(afterParkDelay/2)))
+	return jittered(afterParkDelay)
 }
 
 // resetCurrentBackoff clears the current slot's park escalation.
@@ -177,10 +182,10 @@ func (l *synchronizerList) eligibleCount() int {
 	return count
 }
 
-// earliestUnpark returns the soonest unpark time among the parked slots, and whether there
-// is one. Excluded slots are skipped.
+// earliestUnpark returns the soonest of the unpark times of the parked slots and the end of the
+// FDv1 fallback, and whether there is one. Excluded slots are skipped.
 func (l *synchronizerList) earliestUnpark() (time.Time, bool) {
-	var earliest time.Time
+	earliest := l.fdv1FallbackUntil
 	for i := range l.slots {
 		if l.slots[i].excluded || !l.isParked(i) {
 			continue
@@ -197,20 +202,26 @@ func (l *synchronizerList) hasFDv1Fallback() bool {
 	return l.hasFDv1
 }
 
-// excludeAll excludes every slot and clears the current slot.
-func (l *synchronizerList) excludeAll() {
+// fallBackToFDv1 excludes every FDv2 slot and includes the FDv1 fallback slot until ttl elapses.
+// It also clears the current slot, so the next call to next selects the FDv1 fallback slot. Any
+// park on the FDv1 fallback slot stays in force.
+func (l *synchronizerList) fallBackToFDv1(ttl time.Duration) {
 	for i := range l.slots {
-		l.slots[i].excluded = true
+		l.slots[i].excluded = !l.slots[i].isFDv1Fallback
 	}
 	l.current = -1
+	l.fdv1FallbackUntil = time.Now().Add(ttl)
 }
 
-// fallBackToFDv1 excludes every FDv2 slot and includes the FDv1 fallback slot. It also clears
-// the current slot, so the next call to next selects the FDv1 fallback slot. Any park on the
-// FDv1 fallback slot stays in force.
-func (l *synchronizerList) fallBackToFDv1() {
+// maybeEndFDv1Fallback includes every FDv2 slot and excludes the FDv1 fallback slot if the FDv1
+// fallback TTL has elapsed. It reports whether it ended the fallback.
+func (l *synchronizerList) maybeEndFDv1Fallback() bool {
+	if l.fdv1FallbackUntil.IsZero() || time.Now().Before(l.fdv1FallbackUntil) {
+		return false
+	}
 	for i := range l.slots {
-		l.slots[i].excluded = !l.slots[i].isFDv1Fallback
+		l.slots[i].excluded = l.slots[i].isFDv1Fallback
 	}
-	l.current = -1
+	l.fdv1FallbackUntil = time.Time{}
+	return true
 }

internal/datasystem/synchronizer_list_test.go

@@ -203,7 +203,7 @@ func TestSynchronizerListFDv1FallbackSlotIsExcludedUntilFallback(t *testing.T) {
 	assert.Equal(t, 0, list.currentIndex())
 
 	// Fall back to FDv1 and select again.
-	list.fallBackToFDv1()
+	list.fallBackToFDv1(time.Hour)
 	list.next()
 
 	// Only the FDv1 slot is eligible, and every FDv2 slot is excluded.
@@ -219,7 +219,7 @@ func TestSynchronizerListExcludedSlotIsNotEligibleWhenItsParkExpires(t *testing.
 	// Park the first slot past its unpark time, then engage a directed fallback to FDv1.
 	list.next()
 	list.slots[0].parkedUntil = now.Add(-time.Second)
-	list.fallBackToFDv1()
+	list.fallBackToFDv1(time.Hour)
 
 	// The FDv2 slot is past its unpark time but excluded, so it still does not run.
 	assert.Equal(t, 1, list.eligibleCount())
@@ -228,18 +228,6 @@ func TestSynchronizerListExcludedSlotIsNotEligibleWhenItsParkExpires(t *testing.
 	assert.True(t, list.slots[list.currentIndex()].isFDv1Fallback)
 }
 
-func TestSynchronizerListExcludeAllLeavesNoSlotEligible(t *testing.T) {
-	list := newSynchronizerList([]synchronizerBuilder{noopBuilder, noopBuilder}, noopBuilder)
-	list.next()
-
-	// Exclude every slot, including the FDv1 fallback slot.
-	list.excludeAll()
-
-	// No slot remains eligible.
-	list.next()
-	assert.Equal(t, -1, list.currentIndex())
-	assert.Equal(t, 0, list.eligibleCount())
-}
 
 func TestSynchronizerListUnparkAllKeepsCurrentFDv2Slot(t *testing.T) {
 	// Streaming, polling, and FDv1 fallback slots. Streaming is parked, and polling is current.
@@ -268,7 +256,7 @@ func TestSynchronizerListUnparkAllStaysOnFDv1AfterFallback(t *testing.T) {
 	list.next()
 	list.parkCurrent()
 	list.next()
-	list.fallBackToFDv1()
+	list.fallBackToFDv1(time.Hour)
 	list.next()
 	assert.Equal(t, 2, list.currentIndex())
 

ldclient.go

@@ -710,8 +710,8 @@ func (client *LDClient) IsOffline() bool {
 // the client reconnects with the new key.
 //
 // If the client uses the FDv2 data system, synchronizers that are backing off after a failure can
-// run again at once. If no synchronizer is running, the client starts one with the new key. If the
-// client has fallen back to FDv1, it stays on FDv1.
+// run again at once. If no synchronizer is running, the client starts one with the new key. The key
+// change does not end a fallback to FDv1.
 //
 // The client does not compare sdkKey with the key it uses now. A call with the current key also
 // ends the backoff of every synchronizer. The Relay Proxy uses this to try a key