From bf82b14af439bb1f847814bab5dfb1315139151e Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 6 Nov 2026 18:22:03 +0000 Subject: [PATCH] fix: Fall back from a silent FDv2 synchronizer during startup Before any data has been applied, a synchronizer that provides no data within 10 seconds is abandoned for the next synchronizer, rather than waiting for the 2 minute interrupted fallback. The FDv2 streaming synchronizer now reports Valid after it applies a changeset, so the synchronizer composite can schedule recovery to the primary synchronizer. Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../FDv2DataSources/FDv2DataSource.cs | 129 +++++++++++++++++- .../FDv2StreamingDataSource.cs | 2 + .../FDv2DataSources/FDv2DataSourceTest.cs | 101 ++++++++++++++ .../FDv2StreamingDataSourceTest.cs | 5 + 4 files changed, 234 insertions(+), 3 deletions(-) diff --git a/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2DataSource.cs b/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2DataSource.cs index 94575c09e..02d7f4f1c 100644 --- a/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2DataSource.cs +++ b/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 _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 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 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 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() } + /// + /// Records whether any data has been applied, from any source. + /// + private class DataAppliedObserver : IDataSourceObserver + { + private volatile bool _dataApplied; + + public bool DataApplied => _dataApplied; + + public void Apply(ChangeSet changeSet) + { + _dataApplied = true; + } + + public void UpdateStatus(DataSourceState newState, DataSourceStatus.ErrorInfo? newError) + { + } + } + public enum DataSourceCategory { Initializers, diff --git a/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2StreamingDataSource.cs b/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2StreamingDataSource.cs index 907f0fec1..ef415de2e 100644 --- a/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2StreamingDataSource.cs +++ b/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 { diff --git a/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2DataSourceTest.cs b/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2DataSourceTest.cs index 0f7394cf6..3f8d7e758 100644 --- a/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2DataSourceTest.cs +++ b/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 WaitUntil(Func 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 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 { "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(ChangeSetType.Full, Selector.Make(1, "state"), + new List>>(), 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; } diff --git a/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2StreamingDataSourceTest.cs b/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2StreamingDataSourceTest.cs index 65e47caa2..761be1f34 100644 --- a/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2StreamingDataSourceTest.cs +++ b/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);