diff --git a/pkgs/sdk/server/src/Interfaces/DataSourceStatus.cs b/pkgs/sdk/server/src/Interfaces/DataSourceStatus.cs index d823f0317..260154045 100644 --- a/pkgs/sdk/server/src/Interfaces/DataSourceStatus.cs +++ b/pkgs/sdk/server/src/Interfaces/DataSourceStatus.cs @@ -92,6 +92,12 @@ public struct ErrorInfo /// public bool FDv1Fallback { get; set; } + /// + /// When is true, how long to stay on FDv1 before attempting to + /// recover to FDv2. Null means the directive carried no TTL, so the default applies. + /// + internal TimeSpan? FDv1FallbackTtl { get; set; } + /// /// Constructs an instance based on an exception. /// diff --git a/pkgs/sdk/server/src/Internal/DataSources/CompositeDataSource/CompositeSource.cs b/pkgs/sdk/server/src/Internal/DataSources/CompositeDataSource/CompositeSource.cs index 69b869b54..2c7573b5c 100644 --- a/pkgs/sdk/server/src/Internal/DataSources/CompositeDataSource/CompositeSource.cs +++ b/pkgs/sdk/server/src/Internal/DataSources/CompositeDataSource/CompositeSource.cs @@ -31,6 +31,7 @@ internal sealed class CompositeSource : IDataSource, ICompositeSourceActionable private readonly IDataSourceUpdatesV2 _originalUpdateSink; private readonly IDataSourceUpdatesV2 _sanitizedUpdateSink; private readonly SourcesList<(SourceFactory Factory, ActionApplierFactory ActionApplierFactory, CompositeEntryKind Kind)> _sourcesList; + private readonly List<(SourceFactory Factory, ActionApplierFactory ActionApplierFactory, CompositeEntryKind Kind)> _originalEntries; private readonly DisableableDataSourceUpdatesTracker _disableableTracker; // Tracks the entry from the sources list that was used to create the current @@ -76,6 +77,7 @@ public CompositeSource( // this tracker is used to disconnect the current source from the updates sink when it is no longer needed. _disableableTracker = new DisableableDataSourceUpdatesTracker(); + _originalEntries = new List<(SourceFactory Factory, ActionApplierFactory ActionApplierFactory, CompositeEntryKind Kind)>(factoryTuples); _sourcesList = new SourcesList<(SourceFactory Factory, ActionApplierFactory ActionApplierFactory, CompositeEntryKind Kind)>( circular: circular, initialList: factoryTuples @@ -487,6 +489,23 @@ public void BlockAll(Predicate kindMatches) }); } + public void UnblockAll() + { + if (_disposed) + { + return; + } + + EnqueueAction(() => + { + lock (_lock) + { + _sourcesList.Replace(_originalEntries); + _log.Debug("{0} restored all of its entries.", _compositeDescription); + } + }); + } + private void logTransition(String previousDescription, String currentDescription) { if (previousDescription != null && currentDescription != null) { _log.Debug("{0} transitioned from {1} to {2}.", _compositeDescription, previousDescription, currentDescription); diff --git a/pkgs/sdk/server/src/Internal/DataSources/CompositeDataSource/ICompositeSourceActionable.cs b/pkgs/sdk/server/src/Internal/DataSources/CompositeDataSource/ICompositeSourceActionable.cs index daabaf8a1..9523e78f4 100644 --- a/pkgs/sdk/server/src/Internal/DataSources/CompositeDataSource/ICompositeSourceActionable.cs +++ b/pkgs/sdk/server/src/Internal/DataSources/CompositeDataSource/ICompositeSourceActionable.cs @@ -50,5 +50,12 @@ internal interface ICompositeSourceActionable /// /// predicate selecting which kinds to remove void BlockAll(Predicate kindMatches); + + /// + /// Restores every entry removed by or , so + /// the source list matches its original configuration. The current data source is not + /// disposed -- callers should follow this with . + /// + void UnblockAll(); } } diff --git a/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv1FallbackTtl.cs b/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv1FallbackTtl.cs new file mode 100644 index 000000000..37ba7085d --- /dev/null +++ b/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv1FallbackTtl.cs @@ -0,0 +1,69 @@ +using System; +using System.Collections.Generic; +using System.Globalization; +using System.Linq; + +namespace LaunchDarkly.Sdk.Server.Internal.FDv2DataSources +{ + /// + /// Computes the Fallback TTL of an FDv1 Fallback Directive: how long to stay off FDv2 before + /// attempting to recover to it. + /// + internal static class FDv1FallbackTtl + { + internal const string TtlHeaderName = "x-ld-fd-fallback-ttl"; + + /// + /// Used whenever the service supplies no usable TTL. Also the maximum valid TTL. + /// + internal static readonly TimeSpan DefaultTtl = TimeSpan.FromHours(1); + + private static readonly Random Random = new Random(); + private static readonly object RandomLock = new object(); + + /// + /// Returns the default TTL, jittered by subtracting a value drawn uniformly from + /// [0, DefaultTtl/2], so that SDKs which fell back together do not all retry at once. + /// + internal static TimeSpan Default() + { + double fraction; + lock (RandomLock) + { + fraction = Random.NextDouble(); + } + return DefaultTtl - TimeSpan.FromTicks((long)(DefaultTtl.Ticks / 2 * fraction)); + } + + /// + /// Returns a service-supplied TTL as-is if it is within (0, 1 hour]; otherwise the + /// jittered default, so a fallback is never indefinite. + /// + internal static TimeSpan FromServiceTtl(double? seconds) + { + if (seconds.HasValue && seconds.Value > 0 && seconds.Value <= DefaultTtl.TotalSeconds) + { + return TimeSpan.FromSeconds(seconds.Value); + } + return Default(); + } + + /// + /// Returns the TTL carried by the X-LD-FD-Fallback-TTL header, which is a whole number + /// of seconds. A missing or malformed header gives the jittered default. + /// + internal static TimeSpan FromHeaders(IEnumerable>> headers) + { + var value = headers? + .Where(h => string.Equals(h.Key, TtlHeaderName, StringComparison.OrdinalIgnoreCase)) + .SelectMany(h => h.Value) + .FirstOrDefault(); + if (value != null && + long.TryParse(value.Trim(), NumberStyles.AllowLeadingSign, CultureInfo.InvariantCulture, out var seconds)) + { + return FromServiceTtl(seconds); + } + return Default(); + } + } +} diff --git a/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2ChangeSetTranslator.cs b/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2ChangeSetTranslator.cs index 4e5da6983..434e6304f 100644 --- a/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2ChangeSetTranslator.cs +++ b/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2ChangeSetTranslator.cs @@ -19,13 +19,15 @@ internal static class FDv2ChangeSetTranslator /// Logger for diagnostic messages. /// The environment ID to include in the changeset. /// Whether to mark the changeset as carrying the FDv1 fallback directive. + /// The directive's Fallback TTL, if any. /// A DataStoreTypes.ChangeSet containing the converted data. /// Thrown when the changeset type is unknown. public static DataStoreTypes.ChangeSet ToChangeSet( FDv2ChangeSet changeset, Logger log, string environmentId = null, - bool fdv1Fallback = false) + bool fdv1Fallback = false, + TimeSpan? fdv1FallbackTtl = null) { DataStoreTypes.ChangeSetType changeSetType; switch (changeset.Type) @@ -105,7 +107,8 @@ internal static class FDv2ChangeSetTranslator changeset.Selector, dataBuilder.ToImmutable(), environmentId, - fdv1Fallback); + fdv1Fallback, + fdv1FallbackTtl); } /// diff --git a/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2DataSource.CompletingDataSource.cs b/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2DataSource.CompletingDataSource.cs index 5a4b3c3cf..20a88152a 100644 --- a/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2DataSource.CompletingDataSource.cs +++ b/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2DataSource.CompletingDataSource.cs @@ -1,3 +1,4 @@ +using System; using System.Threading.Tasks; using LaunchDarkly.Sdk.Server.Subsystems; @@ -12,15 +13,18 @@ private class CompletingDataSource : IDataSource { private readonly IDataSource _inner; private readonly InitializationTracker _tracker; + private readonly IDisposable _recovery; - public CompletingDataSource(IDataSource inner, InitializationTracker tracker) + public CompletingDataSource(IDataSource inner, InitializationTracker tracker, IDisposable recovery) { _inner = inner; _tracker = tracker; + _recovery = recovery; } public void Dispose() { + _recovery.Dispose(); _inner.Dispose(); _tracker.Dispose(); } diff --git a/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2DataSource.cs b/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2DataSource.cs index 94575c09e..a4ed73ceb 100644 --- a/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2DataSource.cs +++ b/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2DataSource.cs @@ -46,9 +46,11 @@ public static IDataSource CreateFDv2DataSource( // 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). + // lands on the FDv1 fallback entry (when one is configured). Either way it schedules + // an FDv2 recovery attempt after the directive's Fallback TTL. + var directedFallbackRecovery = new DirectedFallbackRecovery(Any(fdv1Synchronizers), sublogger); ActionApplierFactory fdv1FallbackApplierFactory = - (actionable) => new FDv1FallbackActionApplier(actionable); + (actionable) => new FDv1FallbackActionApplier(actionable, directedFallbackRecovery); var initializationTracker = new InitializationTracker(Any(initializers), Any(synchronizers), Any(fdv1Synchronizers)); @@ -136,7 +138,7 @@ public static IDataSource CreateFDv2DataSource( } return new CompletingDataSource(new CompositeSource("FDv2DataSource", updatesSink, underlyingComposites, sublogger, circular: false), - initializationTracker); + initializationTracker, directedFallbackRecovery); } /// @@ -432,8 +434,9 @@ public void Apply(ChangeSet changeSet) } /// - /// Action applier that observes an FDv1 fallback signal and advances the outer composite - /// to the FDv1 fallback synchronizer entry. The directive may arrive either on + /// Action applier that observes an FDv1 fallback signal, advances the outer composite + /// to the FDv1 fallback synchronizer entry, and schedules FDv2 recovery after the + /// directive's Fallback TTL. The directive may arrive either on /// (errorInfo.FDv1Fallback) or on /// (changeSet.FDv1Fallback) when a successful payload also carried the directive header. /// @@ -441,8 +444,8 @@ public void Apply(ChangeSet changeSet) /// Phase-agnostic: when attached to either the initializers entry or the synchronizers /// entry of the outer FDv2 composite, /// removes every FDv2 entry in one shot, leaving the FDv1 fallback entry (if configured) - /// as the next stop. When no FDv1 fallback entry was configured the outer list is exhausted - /// and the composite halts the data system. + /// as the next stop. When no FDv1 fallback entry was configured, no source runs (the data + /// system reports Interrupted) until the recovery attempt. /// /// The single-shot _triggered flag makes this applier idempotent in case the directive /// arrives more than once on the same propagation chain (for example, on a successful FDv2 @@ -452,26 +455,28 @@ public void Apply(ChangeSet changeSet) internal class FDv1FallbackActionApplier : IDataSourceObserver { private readonly ICompositeSourceActionable _actionable; + private readonly DirectedFallbackRecovery _recovery; private int _triggered; - public FDv1FallbackActionApplier(ICompositeSourceActionable actionable) + public FDv1FallbackActionApplier(ICompositeSourceActionable actionable, DirectedFallbackRecovery recovery) { _actionable = actionable ?? throw new ArgumentNullException(nameof(actionable)); + _recovery = recovery ?? throw new ArgumentNullException(nameof(recovery)); } public void UpdateStatus(DataSourceState newState, DataSourceStatus.ErrorInfo? newError) { if (newError?.FDv1Fallback != true) return; - Trigger(); + Trigger(newError.Value.FDv1FallbackTtl); } public void Apply(ChangeSet changeSet) { if (!changeSet.FDv1Fallback) return; - Trigger(); + Trigger(changeSet.FDv1FallbackTtl); } - private void Trigger() + private void Trigger(TimeSpan? ttl) { if (Interlocked.CompareExchange(ref _triggered, 1, 0) != 0) return; @@ -480,8 +485,102 @@ private void Trigger() // itself down by the time the directive is observed, but DisposeCurrent is // idempotent and guards against sources that don't. _actionable.DisposeCurrent(); - _actionable.GoToNext(); - _actionable.StartCurrent(); + if (_recovery.HasFDv1FallbackSynchronizer) + { + _actionable.GoToNext(); + _actionable.StartCurrent(); + } + + _recovery.Schedule(_actionable, ttl ?? FDv1FallbackTtl.Default()); + } + } + + /// + /// Schedules the FDv2 recovery attempt that ends a Directed Fallback once its Fallback TTL + /// elapses. Shared by every of one FDv2 data source. + /// + /// + /// Recovery restores the FDv2 entries and goes to the first one, re-running the full + /// initialization and synchronization flow. Going to an entry disposes and disconnects the + /// current (FDv1) source first, so FDv1 and FDv2 never write to the store at the same time. + /// A later directive replaces any pending recovery with one using the new TTL. + /// + internal sealed class DirectedFallbackRecovery : IDisposable + { + private readonly object _lock = new object(); + private readonly Logger _log; + private CancellationTokenSource _canceller; + private bool _disposed; + + public bool HasFDv1FallbackSynchronizer { get; } + + public DirectedFallbackRecovery(bool hasFDv1FallbackSynchronizer, Logger logger) + { + HasFDv1FallbackSynchronizer = hasFDv1FallbackSynchronizer; + _log = logger ?? throw new ArgumentNullException(nameof(logger)); + } + + public void Schedule(ICompositeSourceActionable actionable, TimeSpan ttl) + { + CancellationToken token; + lock (_lock) + { + if (_disposed) return; + CancelPendingRecovery(); + _canceller = new CancellationTokenSource(); + token = _canceller.Token; + } + + if (HasFDv1FallbackSynchronizer) + { + _log.Info("Falling back to FDv1; will attempt to recover to FDv2 in {0} seconds.", + ttl.TotalSeconds); + } + else + { + _log.Warn("Received an FDv1 fallback directive, but no FDv1 fallback synchronizer is " + + "configured; will attempt to recover to FDv2 in {0} seconds.", ttl.TotalSeconds); + } + + _ = Task.Run(async () => + { + try + { + await Task.Delay(ttl, token).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + return; + } + + lock (_lock) + { + if (_disposed || token.IsCancellationRequested) return; + } + + _log.Info("FDv1 fallback TTL elapsed; attempting to recover to FDv2."); + actionable.UnblockAll(); + actionable.GoToFirst(); + actionable.StartCurrent(); + }); + } + + public void Dispose() + { + lock (_lock) + { + _disposed = true; + CancelPendingRecovery(); + } + } + + // Must be called while holding _lock. + private void CancelPendingRecovery() + { + if (_canceller == null) return; + _canceller.Cancel(); + _canceller.Dispose(); + _canceller = null; } } diff --git a/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2PollingDataSource.cs b/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2PollingDataSource.cs index 0fae9f5d9..e65166ddf 100644 --- a/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2PollingDataSource.cs +++ b/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2PollingDataSource.cs @@ -100,20 +100,24 @@ private async Task UpdateTaskAsync() ?.FirstOrDefault(); } - var fdv1Fallback = HasFDv1FallbackHeader(response.Value.Headers); - ProcessPollingResponse(response.Value, fdv1Fallback); + var fdv1FallbackTtl = HasFDv1FallbackHeader(response.Value.Headers) + ? FDv1FallbackTtl.FromHeaders(response.Value.Headers) + : (TimeSpan?)null; + var goodbyeFallbackTtl = ProcessPollingResponse(response.Value, fdv1FallbackTtl); + fdv1FallbackTtl = fdv1FallbackTtl ?? goodbyeFallbackTtl; // On a successful response carrying the directive, the changeset above already // rode the FDv1Fallback flag through to the appliers, which trigger the fallback // transition. Tear down the polling loop here so we stop hitting the server. - if (fdv1Fallback) + if (fdv1FallbackTtl.HasValue) { _log.Info("LaunchDarkly polling response indicates fallback to FDv1"); var fallbackError = new DataSourceStatus.ErrorInfo { Kind = DataSourceStatus.ErrorKind.Unknown, Time = DateTime.Now, - FDv1Fallback = true + FDv1Fallback = true, + FDv1FallbackTtl = fdv1FallbackTtl }; // Resolve _initTask so the polling source's Start() task doesn't leak when // ProcessChangeSet failed (e.g. transient data store error): Shutdown below @@ -129,13 +133,10 @@ private async Task UpdateTaskAsync() var recoverable = HttpErrors.IsRecoverable(ex.StatusCode); var errorInfo = DataSourceStatus.ErrorInfo.FromHttpError(ex.StatusCode, recoverable); - // Check for LD fallback header - if (ex.Headers != null) + if (HasFDv1FallbackHeader(ex.Headers)) { - errorInfo.FDv1Fallback = ex.Headers - .Where(h => string.Equals(h.Key, "x-ld-fd-fallback", StringComparison.OrdinalIgnoreCase)) - .SelectMany(h => h.Value) - .Any(v => string.Equals(v, "true", StringComparison.OrdinalIgnoreCase)); + errorInfo.FDv1Fallback = true; + errorInfo.FDv1FallbackTtl = FDv1FallbackTtl.FromHeaders(ex.Headers); } if (errorInfo.Recoverable) @@ -180,8 +181,10 @@ private async Task UpdateTaskAsync() } } - private void ProcessPollingResponse(FDv2PollingResponse response, bool fdv1Fallback) + /// the Fallback TTL of a goodbye that carried an FDv1 Fallback Directive, or null + private TimeSpan? ProcessPollingResponse(FDv2PollingResponse response, TimeSpan? fdv1FallbackTtl) { + TimeSpan? goodbyeFallbackTtl = null; lock (_protocolLock) { _protocolHandler.Reset(); @@ -189,9 +192,14 @@ private void ProcessPollingResponse(FDv2PollingResponse response, bool fdv1Fallb foreach (var evt in response.Events) { var action = _protocolHandler.HandleEvent(evt); - ProcessProtocolAction(action, fdv1Fallback); + if (action is FDv2ActionGoodbye goodbye && goodbye.FDv1FallbackTtl.HasValue) + { + goodbyeFallbackTtl = goodbye.FDv1FallbackTtl; + } + ProcessProtocolAction(action, fdv1FallbackTtl); } } + return goodbyeFallbackTtl; } private void HandleJsonError(string message) @@ -208,12 +216,12 @@ private void HandleJsonError(string message) _dataSourceUpdates.UpdateStatus(DataSourceState.Interrupted, errorInfo); } - private void ProcessProtocolAction(IFDv2ProtocolAction action, bool fdv1Fallback) + private void ProcessProtocolAction(IFDv2ProtocolAction action, TimeSpan? fdv1FallbackTtl) { switch (action) { case FDv2ActionChangeset changesetAction: - ProcessChangeSet(changesetAction.Changeset, fdv1Fallback); + ProcessChangeSet(changesetAction.Changeset, fdv1FallbackTtl); break; case FDv2ActionError errorAction: _log.Error("FDv2 error event: {0} - {1}", errorAction.Id, errorAction.Reason); @@ -241,12 +249,13 @@ private void ProcessProtocolAction(IFDv2ProtocolAction action, bool fdv1Fallback } } - private void ProcessChangeSet(FDv2ChangeSet fdv2ChangeSet, bool fdv1Fallback) + private void ProcessChangeSet(FDv2ChangeSet fdv2ChangeSet, TimeSpan? fdv1FallbackTtl) { if (!(_dataSourceUpdates is ITransactionalDataSourceUpdates transactionalDataSourceUpdates)) throw new InvalidOperationException("Cannot apply updates to non-transactional data source"); - var dataStoreChangeSet = FDv2ChangeSetTranslator.ToChangeSet(fdv2ChangeSet, _log, _environmentId, fdv1Fallback); + var dataStoreChangeSet = FDv2ChangeSetTranslator.ToChangeSet(fdv2ChangeSet, _log, _environmentId, + fdv1FallbackTtl.HasValue, fdv1FallbackTtl); // If the update fails, then we wait until the next poll and try again. // This is different from a streaming data source, which will need to re-start to get an initial diff --git a/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2ProtocolHandler.cs b/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2ProtocolHandler.cs index 5a4d653ab..a6a4d1036 100644 --- a/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2ProtocolHandler.cs +++ b/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2ProtocolHandler.cs @@ -1,3 +1,4 @@ +using System; using System.Collections.Generic; using System.Collections.Immutable; using System.Linq; @@ -101,9 +102,16 @@ internal sealed class FDv2ActionGoodbye : IFDv2ProtocolAction public string Reason { get; } - public FDv2ActionGoodbye(string reason) + /// + /// The Fallback TTL if the goodbye carried an FDv1 Fallback Directive (protocolFallbackTTL), + /// or null if it did not. + /// + public TimeSpan? FDv1FallbackTtl { get; } + + public FDv2ActionGoodbye(string reason, TimeSpan? fdv1FallbackTtl = null) { Reason = reason; + FDv1FallbackTtl = fdv1FallbackTtl; } } @@ -315,7 +323,10 @@ private IFDv2ProtocolAction Error(Error error) /// private IFDv2ProtocolAction Goodbye(Goodbye intent) { - return new FDv2ActionGoodbye(intent.Reason); + return new FDv2ActionGoodbye(intent.Reason, + intent.HasProtocolFallbackTtl + ? FDv1FallbackTtl.FromServiceTtl(intent.ProtocolFallbackTtl) + : (TimeSpan?)null); } /// diff --git a/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2StreamingDataSource.cs b/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2StreamingDataSource.cs index 907f0fec1..d24001335 100644 --- a/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2StreamingDataSource.cs +++ b/pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2StreamingDataSource.cs @@ -53,6 +53,9 @@ internal delegate IEventSource EventSourceCreator(Uri streamUri, // the SDK shuts the stream down and signals the FDv1 fallback action applier. private volatile bool _fdv1FallbackRequested; + // The Fallback TTL of the directive; written before _fdv1FallbackRequested is set. + private TimeSpan _fdv1FallbackTtl; + /// /// When the store enters a failed state, and we don't have "data source monitoring", we want to log /// a message that we are restarting the event source. We don't want to log this message on multiple @@ -227,8 +230,10 @@ private void OnMessage(object sender, MessageReceivedEventArgs e) { var changeset = changeAction.Changeset; var fdv1Fallback = _fdv1FallbackRequested; + TimeSpan? fdv1FallbackTtl = fdv1Fallback ? _fdv1FallbackTtl : (TimeSpan?)null; var storeError = !_transactionalDataSourceUpdates.Apply( - FDv2ChangeSetTranslator.ToChangeSet(changeAction.Changeset, _log, _environmentId, fdv1Fallback)); + FDv2ChangeSetTranslator.ToChangeSet(changeAction.Changeset, _log, _environmentId, + fdv1Fallback, fdv1FallbackTtl)); if (!storeError) { @@ -246,7 +251,8 @@ private void OnMessage(object sender, MessageReceivedEventArgs e) { Kind = DataSourceStatus.ErrorKind.Unknown, Time = DateTime.Now, - FDv1Fallback = true + FDv1Fallback = true, + FDv1FallbackTtl = fdv1FallbackTtl }; Shutdown(fallbackError); return; @@ -264,7 +270,18 @@ private void OnMessage(object sender, MessageReceivedEventArgs e) break; case FDv2ActionGoodbye goodbyeAction: _log.Info(goodbyeAction.Reason); - // TODO: Should we handle this proactively in any way? + if (goodbyeAction.FDv1FallbackTtl.HasValue) + { + _log.Info("LaunchDarkly streaming goodbye indicates fallback to FDv1"); + _initTask.TrySetResult(false); + Shutdown(new DataSourceStatus.ErrorInfo + { + Kind = DataSourceStatus.ErrorKind.Unknown, + Time = DateTime.Now, + FDv1Fallback = true, + FDv1FallbackTtl = goodbyeAction.FDv1FallbackTtl + }); + } break; case FDv2ActionNone _: break; @@ -314,13 +331,10 @@ private void OnError(object sender, ExceptionEventArgs e) var recoverable = HttpErrors.IsRecoverable(status); errorInfo = DataSourceStatus.ErrorInfo.FromHttpError(status, recoverable); - // Check for LD fallback header - if (respEx.Headers != null) + if (FDv2PollingDataSource.HasFDv1FallbackHeader(respEx.Headers)) { - errorInfo.FDv1Fallback = respEx.Headers - .Where(h => string.Equals(h.Key, "x-ld-fd-fallback", StringComparison.OrdinalIgnoreCase)) - .SelectMany(h => h.Value) - .Any(v => string.Equals(v, "true", StringComparison.OrdinalIgnoreCase)); + errorInfo.FDv1Fallback = true; + errorInfo.FDv1FallbackTtl = FDv1FallbackTtl.FromHeaders(respEx.Headers); } RecordStreamInit(true); @@ -371,7 +385,12 @@ private void OnOpen(object sender, StateChangedEventArgs e) // Capture the FDv1 fallback header from the connection-open response. The SDK applies // any payload that arrives on this stream and then shuts the stream down so the action // applier can switch to the FDv1 fallback synchronizer. - _fdv1FallbackRequested = FDv2PollingDataSource.HasFDv1FallbackHeader(e.Headers); + var fdv1FallbackRequested = FDv2PollingDataSource.HasFDv1FallbackHeader(e.Headers); + if (fdv1FallbackRequested) + { + _fdv1FallbackTtl = FDv1FallbackTtl.FromHeaders(e.Headers); + } + _fdv1FallbackRequested = fdv1FallbackRequested; _log.Debug("EventSource Opened"); RecordStreamInit(false); diff --git a/pkgs/sdk/server/src/Internal/FDv2Payloads/Goodbye.cs b/pkgs/sdk/server/src/Internal/FDv2Payloads/Goodbye.cs index 61fd75c3d..23065bccd 100644 --- a/pkgs/sdk/server/src/Internal/FDv2Payloads/Goodbye.cs +++ b/pkgs/sdk/server/src/Internal/FDv2Payloads/Goodbye.cs @@ -15,13 +15,27 @@ internal sealed class Goodbye /// public string Reason { get; } + /// + /// Whether the protocolFallbackTTL field was present. Its presence is an FDv1 Fallback Directive. + /// + public bool HasProtocolFallbackTtl { get; } + + /// + /// The protocolFallbackTTL value in seconds, or null if the field was absent or not a number. + /// + public double? ProtocolFallbackTtl { get; } + /// /// Constructs a new Goodbye. /// /// Reason for the disconnection. - public Goodbye(string reason) + /// Whether the protocolFallbackTTL field was present. + /// The protocolFallbackTTL value in seconds, if it was a number. + public Goodbye(string reason, bool hasProtocolFallbackTtl = false, double? protocolFallbackTtl = null) { Reason = reason; + HasProtocolFallbackTtl = hasProtocolFallbackTtl; + ProtocolFallbackTtl = protocolFallbackTtl; } } @@ -31,12 +45,15 @@ public Goodbye(string reason) internal sealed class GoodbyeConverter : JsonConverter { private const string AttributeReason = "reason"; + private const string AttributeProtocolFallbackTtl = "protocolFallbackTTL"; internal static readonly GoodbyeConverter Instance = new GoodbyeConverter(); public override Goodbye Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options) { string reason = null; + var hasProtocolFallbackTtl = false; + double? protocolFallbackTtl = null; for (var obj = RequireObject(ref reader); obj.Next(ref reader);) { @@ -45,13 +62,24 @@ public override Goodbye Read(ref Utf8JsonReader reader, Type typeToConvert, Json case AttributeReason: reason = reader.GetString(); break; + case AttributeProtocolFallbackTtl: + hasProtocolFallbackTtl = true; + if (reader.TokenType == JsonTokenType.Number) + { + protocolFallbackTtl = reader.GetDouble(); + } + else + { + reader.Skip(); + } + break; default: reader.Skip(); break; } } - return new Goodbye(reason); + return new Goodbye(reason, hasProtocolFallbackTtl, protocolFallbackTtl); } public override void Write(Utf8JsonWriter writer, Goodbye value, JsonSerializerOptions options) @@ -62,6 +90,11 @@ public override void Write(Utf8JsonWriter writer, Goodbye value, JsonSerializerO writer.WriteString(AttributeReason, value.Reason); } + if (value.ProtocolFallbackTtl.HasValue) + { + writer.WriteNumber(AttributeProtocolFallbackTtl, value.ProtocolFallbackTtl.Value); + } + writer.WriteEndObject(); } } diff --git a/pkgs/sdk/server/src/Subsystems/DataStoreTypes.cs b/pkgs/sdk/server/src/Subsystems/DataStoreTypes.cs index f7e322e2d..e47202f13 100644 --- a/pkgs/sdk/server/src/Subsystems/DataStoreTypes.cs +++ b/pkgs/sdk/server/src/Subsystems/DataStoreTypes.cs @@ -436,15 +436,22 @@ public struct ChangeSet /// public bool FDv1Fallback { get; } + /// + /// When is true, how long to stay on FDv1 before attempting to + /// recover to FDv2. Null means the directive carried no TTL, so the default applies. + /// + internal TimeSpan? FDv1FallbackTtl { get; } + internal ChangeSet(ChangeSetType type, Selector selector, IEnumerable>> data, string environmentId, - bool fdv1Fallback = false) + bool fdv1Fallback = false, TimeSpan? fdv1FallbackTtl = null) { Type = type; Selector = selector; Data = data; EnvironmentId = environmentId; FDv1Fallback = fdv1Fallback; + FDv1FallbackTtl = fdv1FallbackTtl; } } } diff --git a/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv1FallbackTtlTest.cs b/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv1FallbackTtlTest.cs new file mode 100644 index 000000000..aba6c6306 --- /dev/null +++ b/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv1FallbackTtlTest.cs @@ -0,0 +1,95 @@ +using System; +using System.Collections.Generic; +using Xunit; + +namespace LaunchDarkly.Sdk.Server.Internal.FDv2DataSources +{ + public class FDv1FallbackTtlTest + { + private static readonly TimeSpan MinDefault = TimeSpan.FromMinutes(30); + private static readonly TimeSpan MaxDefault = TimeSpan.FromHours(1); + + private static void AssertIsDefault(TimeSpan ttl) + { + Assert.InRange(ttl, MinDefault, MaxDefault); + } + + private static List>> Headers(string ttl) => + new List>> + { + new KeyValuePair>("x-ld-fd-fallback", new[] { "true" }), + new KeyValuePair>("X-LD-FD-Fallback-TTL", new[] { ttl }) + }; + + [Fact] + public void DefaultIsJitteredWithinHalfToOneHour() + { + for (var i = 0; i < 100; i++) + { + AssertIsDefault(FDv1FallbackTtl.Default()); + } + } + + [Theory] + [InlineData(1)] + [InlineData(0.5)] + [InlineData(60)] + [InlineData(3600)] + public void ValidServiceTtlIsUsedAsIs(double seconds) + { + Assert.Equal(TimeSpan.FromSeconds(seconds), FDv1FallbackTtl.FromServiceTtl(seconds)); + } + + [Theory] + [InlineData(0)] + [InlineData(-1)] + [InlineData(3601)] + [InlineData(double.NaN)] + [InlineData(double.PositiveInfinity)] + public void InvalidServiceTtlUsesDefault(double seconds) + { + AssertIsDefault(FDv1FallbackTtl.FromServiceTtl(seconds)); + } + + [Fact] + public void MissingServiceTtlUsesDefault() + { + AssertIsDefault(FDv1FallbackTtl.FromServiceTtl(null)); + } + + [Theory] + [InlineData("1", 1)] + [InlineData("60", 60)] + [InlineData(" 120 ", 120)] + [InlineData("3600", 3600)] + public void ValidHeaderTtlIsUsedAsIs(string header, int seconds) + { + Assert.Equal(TimeSpan.FromSeconds(seconds), FDv1FallbackTtl.FromHeaders(Headers(header))); + } + + [Theory] + [InlineData("0")] + [InlineData("-5")] + [InlineData("3601")] + [InlineData("99999999999999999999")] + [InlineData("1.5")] + [InlineData("abc")] + [InlineData("")] + public void InvalidHeaderTtlUsesDefault(string header) + { + AssertIsDefault(FDv1FallbackTtl.FromHeaders(Headers(header))); + } + + [Fact] + public void MissingHeaderTtlUsesDefault() + { + var headers = new List>> + { + new KeyValuePair>("x-ld-fd-fallback", new[] { "true" }) + }; + + AssertIsDefault(FDv1FallbackTtl.FromHeaders(headers)); + AssertIsDefault(FDv1FallbackTtl.FromHeaders(null)); + } + } +} diff --git a/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2DataSourceTest.cs b/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2DataSourceTest.cs index 0f7394cf6..5b2205f4c 100644 --- a/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2DataSourceTest.cs +++ b/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2DataSourceTest.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Threading; using System.Threading.Tasks; using LaunchDarkly.Logging; using LaunchDarkly.Sdk.Server.Interfaces; @@ -1445,13 +1446,21 @@ public void GoToFirst() CallSequence.Add(nameof(GoToFirst)); } + public SemaphoreSlim StartCurrentSignal { get; } = new SemaphoreSlim(0); + public Task StartCurrent() { StartCurrentCalled = true; CallSequence.Add(nameof(StartCurrent)); + StartCurrentSignal.Release(); return Task.FromResult(true); } + public void UnblockAll() + { + CallSequence.Add(nameof(UnblockAll)); + } + public void BlockCurrent() { BlockCurrentCalled = true; @@ -1498,7 +1507,7 @@ public void Reset() public void FDv1FallbackActionApplierIgnoresStatusWithoutFDv1Fallback() { var mockActionable = new MockCompositeSourceActionable(); - var applier = new FDv2DataSource.FDv1FallbackActionApplier(mockActionable); + var applier = new FDv2DataSource.FDv1FallbackActionApplier(mockActionable, new FDv2DataSource.DirectedFallbackRecovery(true, TestLogger)); // Status updates without an FDv1Fallback error must not advance the composite. applier.UpdateStatus(DataSourceState.Off, null); @@ -1519,7 +1528,7 @@ public void FDv1FallbackActionApplierUsesBlockAllAndAdvances() // entry is the next stop), then advances to it. No skip-counting -- the entry list // mutation handles both initializer-phase and synchronizer-phase wirings uniformly. var mockActionable = new MockCompositeSourceActionable(); - var applier = new FDv2DataSource.FDv1FallbackActionApplier(mockActionable); + var applier = new FDv2DataSource.FDv1FallbackActionApplier(mockActionable, new FDv2DataSource.DirectedFallbackRecovery(true, TestLogger)); applier.UpdateStatus( DataSourceState.Off, @@ -1539,7 +1548,7 @@ public void FDv1FallbackActionApplierTriggersOnApplyCarryingDirective() // Successful FDv2 responses that also carry the directive ride the flag on the // ChangeSet rather than UpdateStatus. The applier must trigger from either path. var mockActionable = new MockCompositeSourceActionable(); - var applier = new FDv2DataSource.FDv1FallbackActionApplier(mockActionable); + var applier = new FDv2DataSource.FDv1FallbackActionApplier(mockActionable, new FDv2DataSource.DirectedFallbackRecovery(true, TestLogger)); applier.Apply(new ChangeSet( ChangeSetType.Full, @@ -1560,7 +1569,7 @@ public void FDv1FallbackActionApplierIsIdempotent() // UpdateStatus(Off, FDv1Fallback) during shutdown. The applier must trigger only // once; the second event is a no-op so we don't double-advance off the FDv1 entry. var mockActionable = new MockCompositeSourceActionable(); - var applier = new FDv2DataSource.FDv1FallbackActionApplier(mockActionable); + var applier = new FDv2DataSource.FDv1FallbackActionApplier(mockActionable, new FDv2DataSource.DirectedFallbackRecovery(true, TestLogger)); applier.Apply(new ChangeSet( ChangeSetType.Full, @@ -1576,6 +1585,61 @@ public void FDv1FallbackActionApplierIsIdempotent() Assert.Equal(1, mockActionable.GoToNextCallCount); } + [Fact] + public void FDv1FallbackActionApplierRecoversToFDv2AfterTtl() + { + var mockActionable = new MockCompositeSourceActionable(); + + using (var recovery = new FDv2DataSource.DirectedFallbackRecovery(true, TestLogger)) + { + var applier = new FDv2DataSource.FDv1FallbackActionApplier(mockActionable, recovery); + + applier.UpdateStatus( + DataSourceState.Off, + new DataSourceStatus.ErrorInfo + { + FDv1Fallback = true, + FDv1FallbackTtl = TimeSpan.FromMilliseconds(10) + }); + + // Once for the FDv1 fallback synchronizer, then once for the FDv2 recovery. + Assert.True(mockActionable.StartCurrentSignal.Wait(TimeSpan.FromSeconds(5))); + Assert.True(mockActionable.StartCurrentSignal.Wait(TimeSpan.FromSeconds(5))); + Assert.Equal( + new List + { + "BlockAll", "DisposeCurrent", "GoToNext", "StartCurrent", + "UnblockAll", "GoToFirst", "StartCurrent" + }, + mockActionable.CallSequence); + } + } + + [Fact] + public void FDv1FallbackActionApplierWithoutFDv1SynchronizerRunsNothingThenRecoversToFDv2() + { + var mockActionable = new MockCompositeSourceActionable(); + + using (var recovery = new FDv2DataSource.DirectedFallbackRecovery(false, TestLogger)) + { + var applier = new FDv2DataSource.FDv1FallbackActionApplier(mockActionable, recovery); + + applier.UpdateStatus( + DataSourceState.Off, + new DataSourceStatus.ErrorInfo + { + FDv1Fallback = true, + FDv1FallbackTtl = TimeSpan.FromMilliseconds(10) + }); + + // The only start is the FDv2 recovery; nothing runs while the fallback is engaged. + Assert.True(mockActionable.StartCurrentSignal.Wait(TimeSpan.FromSeconds(5))); + Assert.Equal( + new List { "BlockAll", "DisposeCurrent", "UnblockAll", "GoToFirst", "StartCurrent" }, + mockActionable.CallSequence); + } + } + // End-to-end test: synchronizer reports Off+FDv1Fallback (mirrors the FDv2 streaming // source's Shutdown path on a 403 + x-ld-fd-fallback header). The SDK must engage the // FDv1 fallback synchronizer and reach Initialized via that path. The harness suite @@ -1794,37 +1858,48 @@ public async Task InitializerOffWithFDv1FallbackErrorSkipsSynchronizersAndEngage } } - // End-to-end test: synchronizer reports the FDv1 fallback directive but no FDv1 - // fallback is configured. The SDK must HALT (Requirement 1.6.3(4)) -- which in - // practice means the synchronizer's underlying data source must be disposed so it - // stops trying to reconnect. The data system reaches Off via outer composite - // exhaustion. Mirrors the harness suite "directive without FDv1 fallback configured - // halts the data system". + // End-to-end test: a synchronizer reports the FDv1 directive (500 + header) and no FDv1 + // fallback synchronizer is configured. The data system must stop the synchronizer, report + // Interrupted rather than Off, and restart FDv2 once the Fallback TTL elapses. [Fact] - public void SynchronizerFDv1FallbackWithoutFallbackConfiguredHaltsDataSystem() + public void SynchronizerFDv1FallbackWithoutFallbackConfiguredRecoversToFDv2AfterTtl() { var capturingSink = new CapturingDataSourceUpdatesWithHeaders(); - var synchronizerDisposed = false; + var firstSynchronizerDisposed = false; + var synchronizerStartCount = 0; + var recovered = new TaskCompletionSource(); SourceFactory synchronizerFactory = (updatesSink) => - new HaltMockDataSource( + { + var startNumber = Interlocked.Increment(ref synchronizerStartCount); + return new HaltMockDataSource( onStart: async () => { - // 500 + directive: a recoverable status that would normally drive retries. - // The directive must take precedence and halt those retries. - updatesSink.UpdateStatus( - DataSourceState.Interrupted, - new DataSourceStatus.ErrorInfo - { - Kind = DataSourceStatus.ErrorKind.ErrorResponse, - StatusCode = 500, - FDv1Fallback = true, - Recoverable = true, - Time = DateTime.Now - }); + if (startNumber == 1) + { + updatesSink.UpdateStatus( + DataSourceState.Interrupted, + new DataSourceStatus.ErrorInfo + { + Kind = DataSourceStatus.ErrorKind.ErrorResponse, + StatusCode = 500, + FDv1Fallback = true, + FDv1FallbackTtl = TimeSpan.FromMilliseconds(10), + Recoverable = true, + Time = DateTime.Now + }); + } + else + { + recovered.TrySetResult(firstSynchronizerDisposed); + } await Task.Yield(); }, - onDispose: () => synchronizerDisposed = true); + onDispose: () => + { + if (startNumber == 1) firstSynchronizerDisposed = true; + }); + }; // No FDv1 fallback configured (empty list). var dataSource = FDv2DataSource.CreateFDv2DataSource( @@ -1838,24 +1913,95 @@ public void SynchronizerFDv1FallbackWithoutFallbackConfiguredHaltsDataSystem() { var startTask = dataSource.Start(); - // The data system must transition to Off via outer composite exhaustion. Wait a - // few status updates -- intermediate Interrupted statuses come first. - var statuses = capturingSink.WaitForStatusUpdates(2, TimeSpan.FromSeconds(5)); - Assert.Contains(statuses, s => s.State == DataSourceState.Off); - - // Start() must resolve (with false) rather than hang. Without the - // InitializationTracker's no-fallback shortcut, the tracker would stay in - // FallingBack forever because the outer composite's exhaustion-driven Off goes - // directly to the external sink, bypassing the tracker's observers. + // Start() resolves with false: no data arrived and there is no FDv1 synchronizer + // to take over. Assert.True(startTask.Wait(TimeSpan.FromSeconds(5)), "Start() must resolve when FDv1 directive arrives but no fallback is configured"); - Assert.False(startTask.Result, - "Start() should resolve with false when no FDv1 fallback is configured to take over"); + Assert.False(startTask.Result); + + // FDv2 is restarted after the TTL, and only after the first synchronizer was disposed. + Assert.True(recovered.Task.Wait(TimeSpan.FromSeconds(5)), "FDv2 must recover after the TTL"); + Assert.True(recovered.Task.Result, + "the first synchronizer must be disposed before FDv2 recovers"); + + // The fallback is never terminal, so the data system must not report Off. + Assert.DoesNotContain(capturingSink.GetAllStatusUpdates(), s => s.State == DataSourceState.Off); + } + finally + { + dataSource.Dispose(); + } + } + + // End-to-end test: after the Fallback TTL, the FDv1 fallback synchronizer is stopped before + // FDv2 restarts, so the two never write to the store at the same time. + [Fact] + public void FDv1FallbackSynchronizerIsStoppedBeforeFDv2RecoversAfterTtl() + { + var capturingSink = new CapturingDataSourceUpdatesWithHeaders(); + + var events = new List(); + var recovered = new TaskCompletionSource(); + var synchronizerStartCount = 0; + SourceFactory synchronizerFactory = (updatesSink) => + { + var startNumber = Interlocked.Increment(ref synchronizerStartCount); + return new HaltMockDataSource( + onStart: async () => + { + lock (events) events.Add("fdv2 start " + startNumber); + if (startNumber == 1) + { + updatesSink.UpdateStatus( + DataSourceState.Off, + new DataSourceStatus.ErrorInfo + { + Kind = DataSourceStatus.ErrorKind.ErrorResponse, + StatusCode = 403, + FDv1Fallback = true, + FDv1FallbackTtl = TimeSpan.FromMilliseconds(10), + Time = DateTime.Now + }); + } + else + { + recovered.TrySetResult(true); + } + await Task.Yield(); + }, + onDispose: () => { }); + }; - // Critically, the synchronizer must have been disposed -- this is what stops the - // streaming source from reconnecting in the harness scenario. - Assert.True(synchronizerDisposed, - "synchronizer source must be disposed so the streaming source stops reconnecting"); + SourceFactory fdv1Factory = (updatesSink) => + new HaltMockDataSource( + onStart: async () => + { + lock (events) events.Add("fdv1 start"); + await Task.Yield(); + }, + onDispose: () => + { + lock (events) events.Add("fdv1 dispose"); + }); + + var dataSource = FDv2DataSource.CreateFDv2DataSource( + capturingSink, + new List(), + new List { synchronizerFactory }, + new List { fdv1Factory }, + TestLogger); + + try + { + _ = dataSource.Start(); + + Assert.True(recovered.Task.Wait(TimeSpan.FromSeconds(5)), "FDv2 must recover after the TTL"); + lock (events) + { + Assert.Equal( + new List { "fdv2 start 1", "fdv1 start", "fdv1 dispose", "fdv2 start 2" }, + events); + } } finally { diff --git a/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2PollingDataSourceTest.cs b/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2PollingDataSourceTest.cs index 2e37787dd..a5b9d5d13 100644 --- a/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2PollingDataSourceTest.cs +++ b/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2PollingDataSourceTest.cs @@ -1019,6 +1019,86 @@ public async Task SuccessfulResponseWithFallbackHeaderAppliesPayloadThenSignalsF } } + [Fact] + public async Task SuccessfulResponseWithFallbackTtlHeaderSignalsTtl() + { + var headers = new List>> + { + new KeyValuePair>("x-ld-fd-fallback", new[] { "true" }), + new KeyValuePair>("x-ld-fd-fallback-ttl", new[] { "60" }) + }; + + _mockRequestor.Setup(r => r.PollingRequestAsync(It.IsAny())) + .ReturnsAsync(CreatePollingResponseWithHeaders( + new[] { CreateServerIntentEvent("none", "test-payload", 1) }, + headers)); + + using (var dataSource = MakeDataSource()) + { + var startTask = dataSource.Start(); + + var changeSet = _updateSink.Applies.ExpectValue(); + Assert.True(changeSet.FDv1Fallback); + Assert.Equal(TimeSpan.FromSeconds(60), changeSet.FDv1FallbackTtl); + + var status = _updateSink.StatusUpdates.ExpectValue(); + Assert.Equal(DataSourceState.Off, status.State); + Assert.True(status.LastError.Value.FDv1Fallback); + Assert.Equal(TimeSpan.FromSeconds(60), status.LastError.Value.FDv1FallbackTtl); + + Assert.True(await startTask); + } + } + + [Fact] + public void ErrorResponseWithFallbackTtlHeaderSignalsTtl() + { + var headers = new List>> + { + new KeyValuePair>("x-ld-fd-fallback", new[] { "true" }), + new KeyValuePair>("x-ld-fd-fallback-ttl", new[] { "30" }) + }; + + _mockRequestor.Setup(r => r.PollingRequestAsync(It.IsAny())) + .ThrowsAsync(new UnsuccessfulResponseException(503, headers)); + + using (var dataSource = MakeDataSource()) + { + dataSource.Start(); + + var status = _updateSink.StatusUpdates.ExpectValue(); + Assert.True(status.LastError.Value.FDv1Fallback); + Assert.Equal(TimeSpan.FromSeconds(30), status.LastError.Value.FDv1FallbackTtl); + } + } + + [Fact] + public async Task GoodbyeWithProtocolFallbackTtlAppliesPayloadThenSignalsFDv1Fallback() + { + var goodbye = new FDv2Event(FDv2EventTypes.Goodbye, + JsonDocument.Parse(@"{""reason"":""fallback"",""protocolFallbackTTL"":45}").RootElement); + + _mockRequestor.Setup(r => r.PollingRequestAsync(It.IsAny())) + .ReturnsAsync(CreatePollingResponse( + CreateServerIntentEvent("none", "test-payload", 1), + goodbye)); + + using (var dataSource = MakeDataSource()) + { + var startTask = dataSource.Start(); + + var changeSet = _updateSink.Applies.ExpectValue(); + Assert.Equal(ChangeSetType.None, changeSet.Type); + + var status = _updateSink.StatusUpdates.ExpectValue(); + Assert.Equal(DataSourceState.Off, status.State); + Assert.True(status.LastError.Value.FDv1Fallback); + Assert.Equal(TimeSpan.FromSeconds(45), status.LastError.Value.FDv1FallbackTtl); + + Assert.True(await startTask); + } + } + [Fact] public async Task SuccessfulResponseWithoutFallbackHeaderDoesNotSignalFDv1Fallback() { diff --git a/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2ProtocolHandlerTest.cs b/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2ProtocolHandlerTest.cs index 7fc0a453d..c7427786c 100644 --- a/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2ProtocolHandlerTest.cs +++ b/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2ProtocolHandlerTest.cs @@ -1,3 +1,4 @@ +using System; using System.Collections.Immutable; using System.Text.Json; using LaunchDarkly.Sdk.Server.Internal.FDv2Payloads; @@ -348,6 +349,37 @@ public void Goodbye_ReturnsGoodbyeActionWithReason() Assert.IsType(action); var goodbyeAction = (FDv2ActionGoodbye)action; Assert.Equal("Server is shutting down", goodbyeAction.Reason); + Assert.Null(goodbyeAction.FDv1FallbackTtl); + } + + [Fact] + public void Goodbye_WithValidProtocolFallbackTtl_ReturnsTtlAsIs() + { + var handler = new FDv2ProtocolHandler(); + var goodbyeEvt = new FDv2Event(FDv2EventTypes.Goodbye, + JsonDocument.Parse(@"{""reason"":""fallback"",""protocolFallbackTTL"":120}").RootElement); + + var action = Assert.IsType(handler.HandleEvent(goodbyeEvt)); + + Assert.Equal(TimeSpan.FromSeconds(120), action.FDv1FallbackTtl); + } + + [Theory] + [InlineData("0")] + [InlineData("-1")] + [InlineData("3601")] + [InlineData("\"soon\"")] + [InlineData("null")] + public void Goodbye_WithInvalidProtocolFallbackTtl_ReturnsJitteredDefault(string ttl) + { + var handler = new FDv2ProtocolHandler(); + var goodbyeEvt = new FDv2Event(FDv2EventTypes.Goodbye, + JsonDocument.Parse(@"{""reason"":""fallback"",""protocolFallbackTTL"":" + ttl + "}").RootElement); + + var action = Assert.IsType(handler.HandleEvent(goodbyeEvt)); + + Assert.NotNull(action.FDv1FallbackTtl); + Assert.InRange(action.FDv1FallbackTtl.Value, TimeSpan.FromMinutes(30), TimeSpan.FromHours(1)); } #endregion diff --git a/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2StreamingDataSourceTest.cs b/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2StreamingDataSourceTest.cs index 65e47caa2..452e92cd4 100644 --- a/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2StreamingDataSourceTest.cs +++ b/pkgs/sdk/server/test/Internal/FDv2DataSources/FDv2StreamingDataSourceTest.cs @@ -387,6 +387,86 @@ public void GoodbyeEventLogsInfoWithReason() } } + [Fact] + public void GoodbyeWithoutProtocolFallbackTtlDoesNotSignalFDv1Fallback() + { + using (var dataSource = MakeDataSource()) + { + dataSource.Start(); + + _mockEventSource.TriggerOpen(); + _mockEventSource.TriggerMessage(CreateMessageEvent("goodbye", CreateGoodbyeJson("bye"))); + + // Messages are handled synchronously, so the goodbye has been processed here. + Assert.False(_mockEventSource.IsClosed); + } + } + + [Fact] + public void GoodbyeWithProtocolFallbackTtlSignalsFDv1Fallback() + { + using (var dataSource = MakeDataSource()) + { + dataSource.Start(); + + _mockEventSource.TriggerOpen(); + _mockEventSource.TriggerMessage(CreateMessageEvent("goodbye", + @"{""reason"":""fallback"",""protocolFallbackTTL"":90}")); + + var status = _updateSink.StatusUpdates.ExpectValue(); + Assert.Equal(DataSourceState.Off, status.State); + Assert.True(status.LastError.Value.FDv1Fallback); + Assert.Equal(TimeSpan.FromSeconds(90), status.LastError.Value.FDv1FallbackTtl); + Assert.True(_mockEventSource.IsClosed); + } + } + + [Fact] + public void ErrorResponseWithFallbackTtlHeaderSignalsTtl() + { + using (var dataSource = MakeDataSource()) + { + dataSource.Start(); + + var headers = new List>> + { + new KeyValuePair>("x-ld-fd-fallback", new[] { "true" }), + new KeyValuePair>("x-ld-fd-fallback-ttl", new[] { "0" }) + }; + _mockEventSource.TriggerError(new EventSourceServiceUnsuccessfulResponseException(503, headers)); + + var status = _updateSink.StatusUpdates.ExpectValue(); + Assert.True(status.LastError.Value.FDv1Fallback); + Assert.InRange(status.LastError.Value.FDv1FallbackTtl.Value, + TimeSpan.FromMinutes(30), TimeSpan.FromHours(1)); + } + } + + [Fact] + public void SuccessfulStreamWithFallbackTtlHeaderSignalsTtl() + { + using (var dataSource = MakeDataSource()) + { + dataSource.Start(); + + var headers = new List>> + { + new KeyValuePair>("x-ld-fd-fallback", new[] { "true" }), + new KeyValuePair>("x-ld-fd-fallback-ttl", new[] { "300" }) + }; + _mockEventSource.TriggerOpen(headers); + _mockEventSource.TriggerMessage(CreateMessageEvent("server-intent", + CreateServerIntentJson("none", "p1", 1))); + + var changeSet = _updateSink.Applies.ExpectValue(); + Assert.Equal(TimeSpan.FromSeconds(300), changeSet.FDv1FallbackTtl); + + var status = _updateSink.StatusUpdates.ExpectValue(); + Assert.Equal(DataSourceState.Off, status.State); + Assert.Equal(TimeSpan.FromSeconds(300), status.LastError.Value.FDv1FallbackTtl); + } + } + [Fact] public void HeartbeatEventIsSilentlyIgnored() { diff --git a/pkgs/sdk/server/test/Internal/FDv2Payloads/FDv2PayloadsTest.cs b/pkgs/sdk/server/test/Internal/FDv2Payloads/FDv2PayloadsTest.cs index 3fed34e26..2d67821f5 100644 --- a/pkgs/sdk/server/test/Internal/FDv2Payloads/FDv2PayloadsTest.cs +++ b/pkgs/sdk/server/test/Internal/FDv2Payloads/FDv2PayloadsTest.cs @@ -322,6 +322,40 @@ public void Goodbye_CanDeserializeAndReserialize() var reserialized = JsonSerializer.Serialize(goodbye, GetJsonOptions()); var deserialized2 = JsonSerializer.Deserialize(reserialized, GetJsonOptions()); Assert.Equal("Server is shutting down", deserialized2.Reason); + Assert.False(deserialized2.HasProtocolFallbackTtl); + } + + [Fact] + public void Goodbye_CanDeserializeAndReserializeProtocolFallbackTtl() + { + const string json = @"{ + ""reason"": ""fallback"", + ""protocolFallbackTTL"": 60 + }"; + + var goodbye = JsonSerializer.Deserialize(json, GetJsonOptions()); + + Assert.True(goodbye.HasProtocolFallbackTtl); + Assert.Equal(60, goodbye.ProtocolFallbackTtl); + + var reserialized = JsonSerializer.Serialize(goodbye, GetJsonOptions()); + var deserialized2 = JsonSerializer.Deserialize(reserialized, GetJsonOptions()); + Assert.True(deserialized2.HasProtocolFallbackTtl); + Assert.Equal(60, deserialized2.ProtocolFallbackTtl); + } + + [Fact] + public void Goodbye_NonNumericProtocolFallbackTtlIsPresentWithoutValue() + { + const string json = @"{ + ""reason"": ""fallback"", + ""protocolFallbackTTL"": ""soon"" + }"; + + var goodbye = JsonSerializer.Deserialize(json, GetJsonOptions()); + + Assert.True(goodbye.HasProtocolFallbackTtl); + Assert.Null(goodbye.ProtocolFallbackTtl); } [Fact]