fix: Fall back from a silent FDv2 synchronizer during startup

pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2DataSource.cs

@@ -44,6 +44,14 @@ public static IDataSource CreateFDv2DataSource(
             ActionApplierFactory timedFallbackAndRecoveryApplierFactory =
                 (actionable) => new ActionApplierTimedFallbackAndRecovery(actionable, sublogger);
 
+            // Before any data has been applied, a synchronizer that stays silent is abandoned after a short
+            // startup timeout, as long as there is another synchronizer to fall back to.
+            var dataAppliedObserver = new DataAppliedObserver();
+            ActionApplierFactory synchronizerApplierFactory = synchronizers != null && synchronizers.Count > 1
+                ? (actionable) => new ActionApplierTimedFallbackAndRecovery(actionable, sublogger,
+                    () => dataAppliedObserver.DataApplied)
+                : timedFallbackAndRecoveryApplierFactory;
+
             // The FDv1 fallback applier is phase-agnostic: when the directive is observed it
             // calls BlockAll(FDv2) to remove every FDv2 entry from the outer list, then GoToNext
             // lands on the FDv1 fallback entry (or on exhaustion when none was configured).
@@ -80,6 +88,7 @@ public static IDataSource CreateFDv2DataSource(
                         return new CompositeSource("Initializers", sink, initializerFactory, sublogger, circular: false);
                     },
                     (actionable) => new CompositeObserver(
+                        dataAppliedObserver,
                         initializationObserver,
                         fdv1FallbackApplierFactory(actionable),
                         blacklistWhenSuccessOrOff(actionable)),
@@ -99,13 +108,14 @@ public static IDataSource CreateFDv2DataSource(
                         for (int i = 0; i < synchronizers.Count; i++)
                         {
                             synchronizersFactoryTuples.Add((synchronizers[i],
-                                timedFallbackAndRecoveryApplierFactory,
+                                synchronizerApplierFactory,
                                 CompositeEntryKind.FDv2));
                         }
 
                         return new CompositeSource("Synchronizers", sink, synchronizersFactoryTuples, sublogger);
                     },
-                    (actionable) => new CompositeObserver(synchronizationObserver, fdv1FallbackApplierFactory(actionable)),
+                    (actionable) => new CompositeObserver(dataAppliedObserver, synchronizationObserver,
+                        fdv1FallbackApplierFactory(actionable)),
                     CompositeEntryKind.FDv2
                 ));
             }
@@ -197,22 +207,94 @@ internal class ActionApplierTimedFallbackAndRecovery : IDataSourceObserver
             private CancellationTokenSource _fallbackCanceller;
             private Task _recoveryTask;
             private CancellationTokenSource _recoveryCanceller;
+            private CancellationTokenSource _startupFallbackCanceller;
             private readonly TimeSpan _interruptedFallbackTimeout;
             private readonly TimeSpan _validRecoveryTimeout;
+            private readonly TimeSpan _startupFallbackTimeout;
+            private readonly Func<bool> _hasDataBeenApplied;
             private static readonly TimeSpan DefaultInterruptedFallbackTimeout = TimeSpan.FromMinutes(2);
             private static readonly TimeSpan DefaultValidRecoveryTimeout = TimeSpan.FromMinutes(5);
+            private static readonly TimeSpan DefaultStartupFallbackTimeout = TimeSpan.FromSeconds(10);
 
             public ActionApplierTimedFallbackAndRecovery(ICompositeSourceActionable actionable, Logger logger)
                 : this(actionable, logger, DefaultInterruptedFallbackTimeout, DefaultValidRecoveryTimeout)
             {
             }
 
+            // hasDataBeenApplied reports whether the data system has applied any data. While it has not, the
+            // current source is abandoned if it provides no data within the startup timeout.
+            public ActionApplierTimedFallbackAndRecovery(ICompositeSourceActionable actionable, Logger logger,
+                Func<bool> hasDataBeenApplied)
+                : this(actionable, logger, DefaultInterruptedFallbackTimeout, DefaultValidRecoveryTimeout,
+                    hasDataBeenApplied, DefaultStartupFallbackTimeout)
+            {
+            }
+
             internal ActionApplierTimedFallbackAndRecovery(ICompositeSourceActionable actionable, Logger logger, TimeSpan interruptedFallbackTimeout, TimeSpan validRecoveryTimeout)
+                : this(actionable, logger, interruptedFallbackTimeout, validRecoveryTimeout, null, TimeSpan.Zero)
+            {
+            }
+
+            internal ActionApplierTimedFallbackAndRecovery(ICompositeSourceActionable actionable, Logger logger,
+                TimeSpan interruptedFallbackTimeout, TimeSpan validRecoveryTimeout,
+                Func<bool> hasDataBeenApplied, TimeSpan startupFallbackTimeout)
             {
                 _actionable = actionable ?? throw new ArgumentNullException(nameof(actionable));
                 _log = logger ?? throw new ArgumentNullException(nameof(logger));
                 _interruptedFallbackTimeout = interruptedFallbackTimeout;
                 _validRecoveryTimeout = validRecoveryTimeout;
+                _hasDataBeenApplied = hasDataBeenApplied;
+                _startupFallbackTimeout = startupFallbackTimeout;
+
+                // An applier is created each time its source becomes the current source, so this times
+                // the source from when it is started.
+                if (_hasDataBeenApplied != null && !_hasDataBeenApplied())
+                {
+                    ScheduleStartupFallback();
+                }
+            }
+
+            private void ScheduleStartupFallback()
+            {
+                _startupFallbackCanceller = new CancellationTokenSource();
+                var cancellationToken = _startupFallbackCanceller.Token;
+
+                _ = Task.Run(async () =>
+                {
+                    try
+                    {
+                        await Task.Delay(_startupFallbackTimeout, cancellationToken);
+                    }
+                    catch (OperationCanceledException)
+                    {
+                        return;
+                    }
+
+                    lock (_lock)
+                    {
+                        if (_startupFallbackCanceller == null || cancellationToken.IsCancellationRequested)
+                        {
+                            return;
+                        }
+
+                        _startupFallbackCanceller.Dispose();
+                        _startupFallbackCanceller = null;
+
+                        if (_hasDataBeenApplied())
+                        {
+                            return;
+                        }
+
+                        CancelPendingFallbackTask();
+                        CancelPendingRecoveryTask();
+
+                        _log.Warn("Current data source has not provided data within {0} seconds of starting, falling back to next source.", _startupFallbackTimeout.TotalSeconds);
+
+                        _actionable.DisposeCurrent();
+                        _actionable.GoToNext();
+                        _actionable.StartCurrent();
+                    }
+                }, cancellationToken);
             }
 
             public void UpdateStatus(DataSourceState newState, DataSourceStatus.ErrorInfo? newError)
@@ -228,6 +310,13 @@ public void UpdateStatus(DataSourceState newState, DataSourceStatus.ErrorInfo? n
 
                 lock (_lock)
                 {
+                    // A source that is valid no longer needs the startup fallback, and one that is off falls
+                    // back below.
+                    if (newState == DataSourceState.Valid || newState == DataSourceState.Off)
+                    {
+                        CancelPendingStartupFallback();
+                    }
+
                     // If there's a pending fallback task and status is not Interrupted, cancel it
                     if (_fallbackTask != null && newState != DataSourceState.Interrupted)
                     {
@@ -281,6 +370,7 @@ public void UpdateStatus(DataSourceState newState, DataSourceStatus.ErrorInfo? n
                                     _fallbackCanceller?.Dispose();
                                     _fallbackCanceller = null;
                                     _fallbackTask = null;
+                                    CancelPendingStartupFallback();
 
                                     _log.Warn("Current data source has been interrupted for more than {0} minutes, falling back to next source.", _interruptedFallbackTimeout.TotalMinutes);
 
@@ -367,6 +457,16 @@ private void CancelPendingFallbackTask()
                 _fallbackTask = null;
             }
 
+            private void CancelPendingStartupFallback()
+            {
+                if (_startupFallbackCanceller != null)
+                {
+                    _startupFallbackCanceller.Cancel();
+                    _startupFallbackCanceller.Dispose();
+                    _startupFallbackCanceller = null;
+                }
+            }
+
             private void CancelPendingRecoveryTask()
             {
                 if (_recoveryCanceller != null)
@@ -381,7 +481,11 @@ private void CancelPendingRecoveryTask()
 
             public void Apply(ChangeSet<ItemDescriptor> changeSet)
             {
-                // apply does nothing wrt fallback and recovery, only status matters
+                // Data from the source ends its startup timeout; otherwise only status matters
+                lock (_lock)
+                {
+                    CancelPendingStartupFallback();
+                }
             }
         }
 
@@ -486,6 +590,25 @@ private void Trigger()
         }
 
 
+        /// <summary>
+        /// Records whether any data has been applied, from any source.
+        /// </summary>
+        private class DataAppliedObserver : IDataSourceObserver
+        {
+            private volatile bool _dataApplied;
+
+            public bool DataApplied => _dataApplied;
+
+            public void Apply(ChangeSet<ItemDescriptor> changeSet)
+            {
+                _dataApplied = true;
+            }
+
+            public void UpdateStatus(DataSourceState newState, DataSourceStatus.ErrorInfo? newError)
+            {
+            }
+        }
+
         public enum DataSourceCategory
         {
             Initializers,

pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2StreamingDataSource.cs

@@ -251,6 +251,8 @@ private void OnMessage(object sender, MessageReceivedEventArgs e)
                             Shutdown(fallbackError);
                             return;
                         }
+
+                        _dataSourceUpdates.UpdateStatus(DataSourceState.Valid, null);
                     }
                     else
                     {

pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2DataSourceTest.cs

@@ -1411,6 +1411,107 @@ public async Task FallbackAndRecoveryTasksWellBehaved()
             Assert.False(mockActionable.GoToNextCalled, "GoToNext should not be called when already at first");
         }
 
+        private static async Task<bool> WaitUntil(Func<bool> condition, TimeSpan timeout)
+        {
+            var deadline = DateTime.UtcNow + timeout;
+            while (DateTime.UtcNow < deadline)
+            {
+                if (condition()) return true;
+                await Task.Delay(5);
+            }
+            return condition();
+        }
+
+        private FDv2DataSource.ActionApplierTimedFallbackAndRecovery MakeApplierWithStartupFallback(
+            MockCompositeSourceActionable actionable, Func<bool> hasDataBeenApplied, TimeSpan startupFallbackTimeout) =>
+            new FDv2DataSource.ActionApplierTimedFallbackAndRecovery(
+                actionable,
+                TestLogger,
+                TimeSpan.FromHours(1),
+                TimeSpan.FromHours(1),
+                hasDataBeenApplied,
+                startupFallbackTimeout
+            );
+
+        [Fact]
+        public async Task StartupFallbackFallsBackFromASourceThatProvidesNoData()
+        {
+            var mockActionable = new MockCompositeSourceActionable();
+            MakeApplierWithStartupFallback(mockActionable, () => false, TimeSpan.FromMilliseconds(20));
+
+            Assert.True(await WaitUntil(() => mockActionable.StartCurrentCalled, TimeSpan.FromSeconds(5)),
+                "StartCurrent should be called on startup fallback");
+            Assert.Equal(new List<string> { "DisposeCurrent", "GoToNext", "StartCurrent" },
+                mockActionable.CallSequence);
+            AssertLogMessageRegex(true, LogLevel.Warn,
+                ".*Current data source has not provided data within.*falling back to next source.*");
+        }
+
+        [Fact]
+        public async Task StartupFallbackIsNotScheduledIfDataHasAlreadyBeenApplied()
+        {
+            var mockActionable = new MockCompositeSourceActionable();
+            MakeApplierWithStartupFallback(mockActionable, () => true, TimeSpan.FromMilliseconds(20));
+
+            await Task.Delay(TimeSpan.FromMilliseconds(200));
+
+            Assert.Empty(mockActionable.CallSequence);
+        }
+
+        [Fact]
+        public async Task StartupFallbackIsCancelledWhenTheSourceAppliesData()
+        {
+            var mockActionable = new MockCompositeSourceActionable();
+            var applier = MakeApplierWithStartupFallback(mockActionable, () => false, TimeSpan.FromMilliseconds(100));
+
+            applier.Apply(new ChangeSet<ItemDescriptor>(ChangeSetType.Full, Selector.Make(1, "state"),
+                new List<KeyValuePair<DataKind, KeyedItems<ItemDescriptor>>>(), null));
+
+            await Task.Delay(TimeSpan.FromMilliseconds(300));
+
+            Assert.Empty(mockActionable.CallSequence);
+        }
+
+        [Fact]
+        public async Task StartupFallbackIsCancelledWhenTheSourceIsValid()
+        {
+            var mockActionable = new MockCompositeSourceActionable();
+            mockActionable.SetIsAtFirst(true);
+            var applier = MakeApplierWithStartupFallback(mockActionable, () => false, TimeSpan.FromMilliseconds(100));
+
+            applier.UpdateStatus(DataSourceState.Valid, null);
+
+            await Task.Delay(TimeSpan.FromMilliseconds(300));
+
+            Assert.Empty(mockActionable.CallSequence);
+        }
+
+        [Fact]
+        public async Task StartupFallbackDoesNotFallBackAgainAfterTheSourceIsOff()
+        {
+            var mockActionable = new MockCompositeSourceActionable();
+            var applier = MakeApplierWithStartupFallback(mockActionable, () => false, TimeSpan.FromMilliseconds(100));
+
+            applier.UpdateStatus(DataSourceState.Off, null);
+
+            await Task.Delay(TimeSpan.FromMilliseconds(300));
+
+            Assert.Equal(1, mockActionable.GoToNextCallCount);
+        }
+
+        [Fact]
+        public async Task StartupFallbackStillAppliesWhileTheSourceIsInterrupted()
+        {
+            var mockActionable = new MockCompositeSourceActionable();
+            var applier = MakeApplierWithStartupFallback(mockActionable, () => false, TimeSpan.FromMilliseconds(20));
+
+            applier.UpdateStatus(DataSourceState.Interrupted, null);
+
+            Assert.True(await WaitUntil(() => mockActionable.StartCurrentCalled, TimeSpan.FromSeconds(5)),
+                "StartCurrent should be called on startup fallback");
+            Assert.Equal(1, mockActionable.GoToNextCallCount);
+        }
+
         private class MockCompositeSourceActionable : ICompositeSourceActionable
         {
             public bool DisposeCurrentCalled { get; private set; }

pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2StreamingDataSourceTest.cs

@@ -776,6 +776,10 @@ public async Task SuccessfulStreamWithoutFallbackHeaderDoesNotSignalFDv1Fallback
                 var result = await startTask;
                 Assert.True(result);
 
+                var status = _updateSink.StatusUpdates.ExpectValue();
+                Assert.Equal(DataSourceState.Valid, status.State);
+                Assert.Null(status.LastError);
+
                 _updateSink.StatusUpdates.ExpectNoValue(TimeSpan.FromMilliseconds(50));
                 Assert.False(_mockEventSource.IsClosed);
             }
@@ -840,6 +844,7 @@ public async Task UnrecoverableHttpErrorAfterInitializationReportsOffStatusWithR
 
                 await startTask;
                 Assert.True(dataSource.Initialized);
+                Assert.Equal(DataSourceState.Valid, _updateSink.StatusUpdates.ExpectValue().State);
 
                 // Now trigger an unrecoverable error after initialization
                 var exception = new EventSourceServiceUnsuccessfulResponseException(403);