From d3866c76363e893ecd84b93fdb49d9f199514ec5 Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Fri, 28 Aug 2026 13:06:29 -0400 Subject: [PATCH 1/5] feat!: add temporary retry-delay bounds and apply the SSE retry: field Adds two members to IEventSource so a consumer can slow reconnection at runtime without reconfiguring or recreating the client: void SetTemporaryRetryDelayBounds(TimeSpan initialDelay, TimeSpan maxDelay) void ClearTemporaryRetryDelayBounds() Temporary bounds revert to the configured values automatically once a connection stays open for BackoffResetThreshold, or explicitly via Clear. Reapplying identical bounds is idempotent, so a caller may set them on every failure without pinning the delay at the minimum. The SSE retry: field is now applied to the backoff rather than discarded. A server-directed value replaces the minimum delay, is clamped to one hour, resets the backoff formula input so the next attempt uses the directed value directly, and is sticky: it survives bounds changes and healthy-operation resets until the server sends another value. Ceilings are never affected, so the bound on reconnection keeps applying. Previously the parsed value only acted as an on/off switch, which also meant configuring InitialRetryDelay(TimeSpan.Zero) disabled backoff outright. That gate is gone; backoff is governed by the bounds alone. The delay computation is reworked to integer arithmetic. It previously raised 2 to the power of an unbounded exponent in floating point and clamped the result, which threw OverflowException once the exponent passed 21 with a maximum above int.MaxValue milliseconds -- and that exception escaped the reconnect loop, stopping the stream permanently. It now compares against the ceiling before shifting, so no intermediate can exceed it. One behavior change falls out: bounds below one millisecond now truncate to zero, where the former floating-point path yielded one millisecond at the second attempt. Test coverage includes 85 parity vectors generated from the replaced algorithm, so the rewrite is pinned to the delays the library produced before it. BREAKING CHANGE: IEventSource gains two members. Implementations of the interface must add them; callers of EventSource are unaffected. SDK-2791 --- src/LaunchDarkly.EventSource/Configuration.cs | 4 + .../ConfigurationBuilder.cs | 7 +- src/LaunchDarkly.EventSource/EventSource.cs | 46 +- .../ExponentialBackoffWithDecorrelation.cs | 200 +++++- src/LaunchDarkly.EventSource/IEventSource.cs | 59 +- ...ExponentialBackoffWithDecorrelationTest.cs | 595 +++++++++++++++++- .../ServerDirectedRetryDelayTest.cs | 226 +++++++ .../TemporaryRetryDelayBoundsTest.cs | 337 ++++++++++ 8 files changed, 1422 insertions(+), 52 deletions(-) create mode 100644 test/LaunchDarkly.EventSource.Tests/ServerDirectedRetryDelayTest.cs create mode 100644 test/LaunchDarkly.EventSource.Tests/TemporaryRetryDelayBoundsTest.cs diff --git a/src/LaunchDarkly.EventSource/Configuration.cs b/src/LaunchDarkly.EventSource/Configuration.cs index cbee65d..15840a1 100644 --- a/src/LaunchDarkly.EventSource/Configuration.cs +++ b/src/LaunchDarkly.EventSource/Configuration.cs @@ -65,6 +65,10 @@ public sealed class Configuration /// /// The amount of time a connection must stay open before the EventSource resets its backoff delay. /// + /// + /// A connection that stays open for at least this long also discards any temporary bounds set + /// by . + /// /// public TimeSpan BackoffResetThreshold { get; } diff --git a/src/LaunchDarkly.EventSource/ConfigurationBuilder.cs b/src/LaunchDarkly.EventSource/ConfigurationBuilder.cs index 947b4d4..a91d749 100644 --- a/src/LaunchDarkly.EventSource/ConfigurationBuilder.cs +++ b/src/LaunchDarkly.EventSource/ConfigurationBuilder.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Collections.Generic; using System.Net.Http; using System.Text; @@ -148,6 +148,11 @@ public ConfigurationBuilder MaxRetryDelay(TimeSpan maxRetryDelay) /// value. This prevents long delays from occurring on connections that are only rarely restarted. /// /// + /// A connection that stays open for at least this long also discards any bounds set by + /// , restoring + /// and . + /// + /// /// The default value is . Negative /// values are changed to zero. /// diff --git a/src/LaunchDarkly.EventSource/EventSource.cs b/src/LaunchDarkly.EventSource/EventSource.cs index fa6e540..2491ea3 100644 --- a/src/LaunchDarkly.EventSource/EventSource.cs +++ b/src/LaunchDarkly.EventSource/EventSource.cs @@ -24,8 +24,9 @@ public class EventSource : IEventSource, IDisposable private MemoryStream _eventDataUtf8ByteBuffer; private string _eventName; private string _lastEventId; - private TimeSpan _retryDelay; private readonly ExponentialBackoffWithDecorrelation _backOff; + private readonly TimeSpan _configuredInitialRetryDelay; + private readonly TimeSpan _configuredMaxRetryDelay; private CancellationTokenSource _currentRequestToken; private DateTime? _lastSuccessfulConnectionTime; private ReadyState _readyState; @@ -78,6 +79,9 @@ internal TimeSpan BackOffDelay private set; } + // Exposed for tests, so that retry-bounds behavior can be asserted directly + internal ExponentialBackoffWithDecorrelation BackOff => _backOff; + #endregion #region Public Constructors @@ -94,9 +98,11 @@ public EventSource(Configuration configuration) _logger = _configuration.Logger; - _retryDelay = _configuration.InitialRetryDelay; + _configuredInitialRetryDelay = _configuration.InitialRetryDelay; + _configuredMaxRetryDelay = _configuration.MaxRetryDelay; - _backOff = new ExponentialBackoffWithDecorrelation(_retryDelay, _configuration.MaxRetryDelay); + _backOff = new ExponentialBackoffWithDecorrelation(_configuredInitialRetryDelay, + _configuredMaxRetryDelay); _httpClient = _configuration.HttpClient ?? CreateHttpClient(); } @@ -125,7 +131,11 @@ public async Task StartAsync() { if (DateTime.Now.Subtract(_lastSuccessfulConnectionTime.Value) >= _configuration.BackoffResetThreshold) { - _backOff.ResetReconnectAttemptCount(); + // Sustained healthy operation reverts any temporary bounds and resets + // n. Reverting takes precedence over bounds that were + // set during the preceding fault window. + ClearTemporaryRetryDelayBounds(); + _backOff.ResetBackoffN(); } _lastSuccessfulConnectionTime = null; } @@ -199,14 +209,12 @@ public async Task StartAsync() } private async Task MaybeWaitWithBackOff() { - if (_retryDelay.TotalMilliseconds > 0) + TimeSpan sleepTime = _backOff.GetNextBackOff(); + if (sleepTime > TimeSpan.Zero) { - TimeSpan sleepTime = _backOff.GetNextBackOff(); - if (sleepTime.TotalMilliseconds > 0) { - _logger.Info("Waiting {0} milliseconds before reconnecting...", sleepTime.TotalMilliseconds); - BackOffDelay = sleepTime; - await Task.Delay(sleepTime); - } + _logger.Info("Waiting {0} milliseconds before reconnecting...", sleepTime.TotalMilliseconds); + BackOffDelay = sleepTime; + await Task.Delay(sleepTime); } } @@ -221,12 +229,24 @@ public void Restart(bool resetBackoffDelay) } if (resetBackoffDelay) { - _backOff.ResetReconnectAttemptCount(); + _backOff.ResetBackoffN(); } } CancelCurrentRequest(); } + /// + public void SetTemporaryRetryDelayBounds(TimeSpan initialDelay, TimeSpan maxDelay) + { + _backOff.SetBounds(initialDelay, maxDelay); + } + + /// + public void ClearTemporaryRetryDelayBounds() + { + _backOff.SetBounds(_configuredInitialRetryDelay, _configuredMaxRetryDelay); + } + /// /// Closes the connection to the SSE server. The EventSource cannot be reopened after this. /// @@ -411,7 +431,7 @@ private void HandleParsedLine(EventParser.Result result) { if (long.TryParse(result.GetValueAsString(), out var retry)) { - _retryDelay = TimeSpan.FromMilliseconds(retry); + _backOff.SetServerDirectedMinDelay(TimeSpan.FromMilliseconds(retry)); } } } diff --git a/src/LaunchDarkly.EventSource/ExponentialBackoffWithDecorrelation.cs b/src/LaunchDarkly.EventSource/ExponentialBackoffWithDecorrelation.cs index cc6d559..5c8bddf 100644 --- a/src/LaunchDarkly.EventSource/ExponentialBackoffWithDecorrelation.cs +++ b/src/LaunchDarkly.EventSource/ExponentialBackoffWithDecorrelation.cs @@ -2,47 +2,203 @@ namespace LaunchDarkly.EventSource { - internal class ExponentialBackoffWithDecorrelation + internal sealed class ExponentialBackoffWithDecorrelation { - private readonly TimeSpan _minimumDelay; - private readonly TimeSpan _maximumDelay; - private readonly Random _jitterer = new Random(); - private int _reconnectAttempts; + internal static readonly TimeSpan MaxServerDirectedMinDelay = TimeSpan.FromHours(1); + + private readonly object _lock = new object(); + private readonly Random _randomSource; + + private TimeSpan _minimumDelay; + private TimeSpan _maximumDelay; + private int _backoffN; + private TimeSpan? _serverDirectedMinDelay; public ExponentialBackoffWithDecorrelation(TimeSpan minimumDelay, TimeSpan maximumDelay) + : this(minimumDelay, maximumDelay, new Random()) { - _minimumDelay = minimumDelay; - _maximumDelay = maximumDelay; } /// - /// Gets the next backoff duration and increments the reconnect attempt count + /// Constructs an instance with a caller-supplied , so that tests can + /// make jitter deterministic. + /// + internal ExponentialBackoffWithDecorrelation(TimeSpan minimumDelay, TimeSpan maximumDelay, + Random randomSource) + { + _minimumDelay = NonNegative(minimumDelay); + _maximumDelay = NonNegative(maximumDelay); + _randomSource = randomSource ?? new Random(); + } + + /// + /// Gets the next backoff duration, advancing n. /// public TimeSpan GetNextBackOff() { - int nextDelay = GetMaximumMillisecondsForAttempt(_reconnectAttempts); - nextDelay = nextDelay / 2 + _jitterer.Next(nextDelay) / 2; - _reconnectAttempts++; - return TimeSpan.FromMilliseconds(nextDelay); + lock (_lock) + { + long unjittered = GetUnjitteredMillisecondsForN(_backoffN); + + // Random.Next takes an int bound. 2^31 milliseconds is far longer than any + // reconnect delay we would use, so saturating here cannot affect a realistic + // delay. + int jitterBound = unjittered > int.MaxValue ? int.MaxValue : (int)unjittered; + long delayMillis = unjittered / 2 + + (jitterBound > 0 ? _randomSource.Next(jitterBound) / 2 : 0); + + _backoffN++; + return TimeSpan.FromMilliseconds(delayMillis); + } + } + + /// + /// Returns the delay in whole milliseconds for a given backoff n, before jitter is + /// applied: minimumDelay * 2^n, limited to the maximum. + /// + /// + /// n is compared against the ceiling rather than used to compute a value that is + /// then clamped, so no intermediate ever exceeds the maximum. That matters because + /// n is unbounded: a long outage keeps incrementing it, and computing + /// minimumDelay * 2^n directly would overflow. + /// + internal long GetUnjitteredMillisecondsForN(int n) + { + long min = (_serverDirectedMinDelay ?? _minimumDelay).Ticks + / TimeSpan.TicksPerMillisecond; + long max = _maximumDelay.Ticks / TimeSpan.TicksPerMillisecond; + + if (min <= 0 || max <= 0) + { + return 0; + } + if (n <= 0) + { + return min > max ? max : min; + } + + // Shifting the ceiling down cannot overflow, whereas shifting the minimum up can. + // The n >= 63 test is required rather than defensive: C# masks a 64-bit shift count + // to its low 6 bits, so `max >> 64` would silently mean `max >> 0`. + if (n >= 63 || min > (max >> n)) + { + return max; + } + return min << n; + } + + /// + /// Applies a server-directed reconnection time received via the SSE retry: field, + /// replacing the minimum delay used to compute subsequent backoffs. + /// + /// + /// + /// The override takes precedence over both the configured minimum and any minimum + /// installed by . It is sticky: it survives bounds changes and + /// , and is replaced only by a later server-directed value. + /// + /// + /// The maximum is never affected, so the ceiling that bounds reconnection keeps applying. + /// + /// + /// n resets, so the next attempt uses the directed value itself rather than continuing to + /// double from wherever the progression had reached. + /// + /// + /// Values above are capped at it. Negative + /// values become zero. + /// + /// + /// the server-directed reconnection time + public void SetServerDirectedMinDelay(TimeSpan minDelay) + { + var value = NonNegative(minDelay); + if (value > MaxServerDirectedMinDelay) + { + value = MaxServerDirectedMinDelay; + } + lock (_lock) + { + _serverDirectedMinDelay = value; + _backoffN = 0; + } + } + + /// + /// Replaces the delay bounds. + /// + /// + /// If the new bounds differ from the current ones, n resets so that the + /// next delay starts from the new minimum. If they are the same, this does nothing, which + /// makes repeated calls with identical values idempotent -- a caller that reapplies the + /// same bounds on every failure will not pin the delay at the minimum. + /// + /// the new minimum delay; negative values become zero + /// the new maximum delay; negative values become zero + public void SetBounds(TimeSpan minimumDelay, TimeSpan maximumDelay) + { + var min = NonNegative(minimumDelay); + var max = NonNegative(maximumDelay); + lock (_lock) + { + if (min == _minimumDelay && max == _maximumDelay) + { + return; + } + _minimumDelay = min; + _maximumDelay = max; + _backoffN = 0; + } + } + + /// + /// Resets n so the next delay starts from the current minimum, without + /// changing the bounds. + /// + public void ResetBackoffN() + { + lock (_lock) + { + _backoffN = 0; + } } - internal int GetMaximumMillisecondsForAttempt(int attempt) + // Accessors for tests. + + internal int GetBackoffN() { - return Convert.ToInt32(Math.Min(_maximumDelay.TotalMilliseconds, - _minimumDelay.TotalMilliseconds * Math.Pow(2, attempt))); + lock (_lock) + { + return _backoffN; + } } - public int GetReconnectAttemptCount() { - return _reconnectAttempts; + internal TimeSpan? GetServerDirectedMinDelay() + { + lock (_lock) + { + return _serverDirectedMinDelay; + } } - public void ResetReconnectAttemptCount() { - _reconnectAttempts = 0; + internal TimeSpan GetMinimumDelay() + { + lock (_lock) + { + return _minimumDelay; + } } - [Obsolete("IncrementReconnectAttemptCount is deprecated, use GetNextBackOff instead.")] - public void IncrementReconnectAttemptCount() { - _reconnectAttempts++; + internal TimeSpan GetMaximumDelay() + { + lock (_lock) + { + return _maximumDelay; + } } + + // Matches the clamping behavior of ConfigurationBuilder's retry-delay setters: a negative + // duration becomes zero rather than throwing. + private static TimeSpan NonNegative(TimeSpan t) => t < TimeSpan.Zero ? TimeSpan.Zero : t; } } diff --git a/src/LaunchDarkly.EventSource/IEventSource.cs b/src/LaunchDarkly.EventSource/IEventSource.cs index fbeb03f..4fd0cf3 100644 --- a/src/LaunchDarkly.EventSource/IEventSource.cs +++ b/src/LaunchDarkly.EventSource/IEventSource.cs @@ -70,10 +70,67 @@ public interface IEventSource /// /// /// true if the delay before reconnection should be reset to - /// the lowest level (); false if it + /// the lowest level of the currently active bounds, which could be the temporary bounds if + /// is in effect; false if it /// should increase according to the usual exponential backoff logic void Restart(bool resetBackoffDelay); + /// + /// Temporarily replaces the bounds used to compute reconnection delays. + /// + /// + /// + /// The configured and + /// normally govern the + /// exponential backoff between reconnection attempts. This method installs different + /// bounds at runtime, so a caller can slow reconnection down without reconfiguring or + /// recreating the EventSource. + /// + /// + /// The bounds are temporary: they remain in effect until either a connection stays + /// open for at least + /// , at which point the + /// configured bounds are restored automatically, or + /// is called. + /// + /// + /// Changing the bounds resets the backoff level, so the next delay is computed from the new + /// rather than continuing to grow from where it left off. + /// Calling this with bounds that are already in effect does nothing, so it is safe to call + /// repeatedly. + /// + /// + /// Negative values are changed to zero. If is greater than + /// , delays are limited to . No + /// argument causes this method to throw, and no argument can cause a later reconnect to + /// fail. + /// + /// + /// It is safe to call this from any thread, including from an + /// handler. + /// + /// + /// the lowest delay to use while these bounds are in effect + /// the highest delay to use while these bounds are in effect + /// + void SetTemporaryRetryDelayBounds(TimeSpan initialDelay, TimeSpan maxDelay); + + /// + /// Discards any bounds installed by + /// , restoring the configured + /// and + /// . + /// + /// + /// Restoring the configured bounds resets the backoff level. If no temporary bounds + /// are in effect this does nothing. The count of reconnection attempts is not reset. + /// Calling this is not required as the configured bounds are also restored automatically + /// after a connection stays open for + /// . + /// + /// + void ClearTemporaryRetryDelayBounds(); + /// /// Closes the connection to the SSE server. The EventSource cannot be reopened after this. /// diff --git a/test/LaunchDarkly.EventSource.Tests/ExponentialBackoffWithDecorrelationTest.cs b/test/LaunchDarkly.EventSource.Tests/ExponentialBackoffWithDecorrelationTest.cs index 071b119..90be714 100644 --- a/test/LaunchDarkly.EventSource.Tests/ExponentialBackoffWithDecorrelationTest.cs +++ b/test/LaunchDarkly.EventSource.Tests/ExponentialBackoffWithDecorrelationTest.cs @@ -5,51 +5,616 @@ namespace LaunchDarkly.EventSource.Tests { public class ExponentialBackoffWithDecorrelationTest { + private static readonly TimeSpan Base1s = TimeSpan.FromMilliseconds(1000); + private static readonly TimeSpan Max30s = TimeSpan.FromMilliseconds(30000); + + private static ExponentialBackoffWithDecorrelation Seeded(TimeSpan min, TimeSpan max, + int seed = 12345) => + new ExponentialBackoffWithDecorrelation(min, max, new Random(seed)); + [Fact] public void Exponential_backoff_should_not_exceed_maximum() { - TimeSpan max = TimeSpan.FromMilliseconds(30000); ExponentialBackoffWithDecorrelation expo = - new ExponentialBackoffWithDecorrelation(TimeSpan.FromMilliseconds(1000), max); + new ExponentialBackoffWithDecorrelation(Base1s, Max30s); var backoff = expo.GetNextBackOff(); - Assert.True(backoff <= max); + Assert.True(backoff <= Max30s); } [Fact] public void Exponential_backoff_should_not_exceed_maximum_in_test_loop() { - TimeSpan max = TimeSpan.FromMilliseconds(30000); - ExponentialBackoffWithDecorrelation expo = - new ExponentialBackoffWithDecorrelation(TimeSpan.FromMilliseconds(1000), max); + new ExponentialBackoffWithDecorrelation(Base1s, Max30s); for (int i = 0; i < 100; i++) { - var backoff = expo.GetNextBackOff(); - Assert.True(backoff <= max); + Assert.True(backoff <= Max30s); } - } [Fact] - public void Exponential_backoff_should_reset_when_reconnect_count_resets() + public void Exponential_backoff_should_reset_when_backoff_n_resets() { - TimeSpan max = TimeSpan.FromMilliseconds(30000); - ExponentialBackoffWithDecorrelation expo = - new ExponentialBackoffWithDecorrelation(TimeSpan.FromMilliseconds(1000), max); + new ExponentialBackoffWithDecorrelation(Base1s, Max30s); for (int i = 0; i < 100; i++) { - var backoff = expo.GetNextBackOff(); + expo.GetNextBackOff(); } - expo.ResetReconnectAttemptCount(); + expo.ResetBackoffN(); // Backoffs use jitter, so assert that the reset backoff time isn't more than double the minimum Assert.True(expo.GetNextBackOff() <= TimeSpan.FromMilliseconds(2000)); } + + // ---- parity with the previous implementation ---- + + // These expectations were generated by running the algorithm this class replaced -- + // Convert.ToInt32(Math.Min(max.TotalMilliseconds, min.TotalMilliseconds * Math.Pow(2, level))) + // -- so the rewrite is pinned to the delays the library produced before it. The former + // algorithm threw OverflowException for a maximum above int.MaxValue milliseconds; those + // inputs have no previous behavior to match and are covered separately below. + [Theory] + // library defaults + [InlineData(1000, 30000, 0, 1000L)] + [InlineData(1000, 30000, 1, 2000L)] + [InlineData(1000, 30000, 2, 4000L)] + [InlineData(1000, 30000, 3, 8000L)] + [InlineData(1000, 30000, 4, 16000L)] + [InlineData(1000, 30000, 5, 30000L)] + [InlineData(1000, 30000, 6, 30000L)] + [InlineData(1000, 30000, 7, 30000L)] + [InlineData(1000, 30000, 8, 30000L)] + // low ceiling + [InlineData(1000, 3000, 0, 1000L)] + [InlineData(1000, 3000, 1, 2000L)] + [InlineData(1000, 3000, 2, 3000L)] + [InlineData(1000, 3000, 3, 3000L)] + [InlineData(1000, 3000, 4, 3000L)] + // min above max + [InlineData(5000, 1000, 0, 1000L)] + [InlineData(5000, 1000, 1, 1000L)] + [InlineData(5000, 1000, 2, 1000L)] + // min equals max + [InlineData(1000, 1000, 0, 1000L)] + [InlineData(1000, 1000, 1, 1000L)] + [InlineData(1000, 1000, 2, 1000L)] + // ceiling is an exact power-of-two multiple + [InlineData(1000, 32000, 0, 1000L)] + [InlineData(1000, 32000, 1, 2000L)] + [InlineData(1000, 32000, 2, 4000L)] + [InlineData(1000, 32000, 3, 8000L)] + [InlineData(1000, 32000, 4, 16000L)] + [InlineData(1000, 32000, 5, 32000L)] + [InlineData(1000, 32000, 6, 32000L)] + // zero min + [InlineData(0, 30000, 0, 0L)] + [InlineData(0, 30000, 1, 0L)] + [InlineData(0, 30000, 2, 0L)] + // zero max + [InlineData(1000, 0, 0, 0L)] + [InlineData(1000, 0, 1, 0L)] + [InlineData(1000, 0, 2, 0L)] + // one millisecond + [InlineData(1, 1, 0, 1L)] + [InlineData(1, 1, 1, 1L)] + [InlineData(1, 1, 2, 1L)] + // temporary bounds used in tests + [InlineData(60, 500, 0, 60L)] + [InlineData(60, 500, 1, 120L)] + [InlineData(60, 500, 2, 240L)] + [InlineData(60, 500, 3, 480L)] + [InlineData(60, 500, 4, 500L)] + [InlineData(60, 500, 5, 500L)] + // extended regime: 5 min to 1 hr + [InlineData(300000, 3600000, 0, 300000L)] + [InlineData(300000, 3600000, 1, 600000L)] + [InlineData(300000, 3600000, 2, 1200000L)] + [InlineData(300000, 3600000, 3, 2400000L)] + [InlineData(300000, 3600000, 4, 3600000L)] + [InlineData(300000, 3600000, 5, 3600000L)] + [InlineData(300000, 3600000, 6, 3600000L)] + // ceiling at int.MaxValue ms + [InlineData(1000, 2147483647, 0, 1000L)] + [InlineData(1000, 2147483647, 1, 2000L)] + [InlineData(1000, 2147483647, 2, 4000L)] + [InlineData(1000, 2147483647, 3, 8000L)] + [InlineData(1000, 2147483647, 4, 16000L)] + [InlineData(1000, 2147483647, 5, 32000L)] + [InlineData(1000, 2147483647, 6, 64000L)] + [InlineData(1000, 2147483647, 7, 128000L)] + [InlineData(1000, 2147483647, 8, 256000L)] + [InlineData(1000, 2147483647, 9, 512000L)] + [InlineData(1000, 2147483647, 10, 1024000L)] + [InlineData(1000, 2147483647, 11, 2048000L)] + [InlineData(1000, 2147483647, 12, 4096000L)] + [InlineData(1000, 2147483647, 13, 8192000L)] + [InlineData(1000, 2147483647, 14, 16384000L)] + [InlineData(1000, 2147483647, 15, 32768000L)] + [InlineData(1000, 2147483647, 16, 65536000L)] + [InlineData(1000, 2147483647, 17, 131072000L)] + [InlineData(1000, 2147483647, 18, 262144000L)] + [InlineData(1000, 2147483647, 19, 524288000L)] + [InlineData(1000, 2147483647, 20, 1048576000L)] + [InlineData(1000, 2147483647, 21, 2097152000L)] + [InlineData(1000, 2147483647, 22, 2147483647L)] + [InlineData(1000, 2147483647, 23, 2147483647L)] + [InlineData(1000, 2147483647, 24, 2147483647L)] + public void MatchesPreviousImplementation(long minMillis, long maxMillis, int level, + long expected) + { + var expo = Seeded(TimeSpan.FromMilliseconds(minMillis), + TimeSpan.FromMilliseconds(maxMillis)); + + Assert.Equal(expected, expo.GetUnjitteredMillisecondsForN(level)); + } + + // ---- delay math ---- + + [Fact] + public void DelayDoublesPerN() + { + var expo = Seeded(Base1s, Max30s); + + Assert.Equal(1000L, expo.GetUnjitteredMillisecondsForN(0)); + Assert.Equal(2000L, expo.GetUnjitteredMillisecondsForN(1)); + Assert.Equal(4000L, expo.GetUnjitteredMillisecondsForN(2)); + Assert.Equal(8000L, expo.GetUnjitteredMillisecondsForN(3)); + Assert.Equal(16000L, expo.GetUnjitteredMillisecondsForN(4)); + } + + [Fact] + public void DelayClampsAtMaximum() + { + var expo = Seeded(Base1s, TimeSpan.FromMilliseconds(3000)); + + Assert.Equal(1000L, expo.GetUnjitteredMillisecondsForN(0)); + Assert.Equal(2000L, expo.GetUnjitteredMillisecondsForN(1)); + Assert.Equal(3000L, expo.GetUnjitteredMillisecondsForN(2)); + Assert.Equal(3000L, expo.GetUnjitteredMillisecondsForN(3)); + } + + [Fact] + public void DelayReachesMaximumExactlyWhenItIsAPowerOfTwoMultiple() + { + // The boundary the ceiling comparison has to get right: min << level == max exactly. + var expo = Seeded(Base1s, TimeSpan.FromMilliseconds(32000)); + + Assert.Equal(16000L, expo.GetUnjitteredMillisecondsForN(4)); + Assert.Equal(32000L, expo.GetUnjitteredMillisecondsForN(5)); + Assert.Equal(32000L, expo.GetUnjitteredMillisecondsForN(6)); + } + + [Fact] + public void DelayOneBelowAPowerOfTwoCeilingIsCapped() + { + var expo = Seeded(Base1s, TimeSpan.FromMilliseconds(31999)); + + Assert.Equal(16000L, expo.GetUnjitteredMillisecondsForN(4)); + Assert.Equal(31999L, expo.GetUnjitteredMillisecondsForN(5)); + } + + [Fact] + public void DelayWithZeroBaseIsAlwaysZero() + { + var expo = Seeded(TimeSpan.Zero, Max30s); + + Assert.Equal(0L, expo.GetUnjitteredMillisecondsForN(0)); + Assert.Equal(0L, expo.GetUnjitteredMillisecondsForN(5)); + } + + [Fact] + public void DelayWithZeroMaximumIsAlwaysZero() + { + var expo = Seeded(Base1s, TimeSpan.Zero); + + Assert.Equal(0L, expo.GetUnjitteredMillisecondsForN(0)); + Assert.Equal(0L, expo.GetUnjitteredMillisecondsForN(5)); + } + + [Fact] + public void DelayWithBaseAboveMaximumIsClampedToMaximum() + { + var expo = Seeded(TimeSpan.FromMilliseconds(5000), TimeSpan.FromMilliseconds(1000)); + + Assert.Equal(1000L, expo.GetUnjitteredMillisecondsForN(0)); + Assert.Equal(1000L, expo.GetUnjitteredMillisecondsForN(3)); + } + + [Theory] + [InlineData(0)] + [InlineData(1)] + [InlineData(62)] + [InlineData(63)] + [InlineData(64)] + [InlineData(1000)] + [InlineData(int.MaxValue)] + public void HighBackoffNDoesNotOverflow(int level) + { + // Levels at and beyond 63 matter specifically: C# masks a 64-bit shift count to its + // low 6 bits, so a missing guard would silently produce a wrong value rather than + // fail loudly. + var expo = Seeded(Base1s, Max30s); + + Assert.InRange(expo.GetUnjitteredMillisecondsForN(level), 1000L, 30000L); + } + + [Theory] + [InlineData(0)] + [InlineData(22)] + [InlineData(62)] + [InlineData(63)] + [InlineData(64)] + [InlineData(int.MaxValue)] + public void HugeMaximumNeverOverflows(int level) + { + // The previous implementation threw OverflowException here once the level passed 21, + // and that exception surfaced outside the reconnect loop's try/catch, permanently + // stopping the stream. + var expo = Seeded(Base1s, TimeSpan.MaxValue); + + Assert.True(expo.GetUnjitteredMillisecondsForN(level) >= 1000L); + } + + [Fact] + public void NegativeNIsTreatedAsZero() + { + var expo = Seeded(Base1s, Max30s); + + Assert.Equal(1000L, expo.GetUnjitteredMillisecondsForN(-1)); + } + + [Fact] + public void SubMillisecondBoundsTruncateToZero() + { + // Delays are whole milliseconds, so a bound below 1ms rounds down to nothing. The + // previous implementation kept fractional milliseconds in a double and would have + // produced 1ms at level 1 from a 0.5ms minimum. + var expo = Seeded(TimeSpan.FromTicks(5000), TimeSpan.FromMilliseconds(1000)); + + Assert.Equal(0L, expo.GetUnjitteredMillisecondsForN(0)); + Assert.Equal(0L, expo.GetUnjitteredMillisecondsForN(1)); + } + + [Fact] + public void MaximumAboveIntRangeIsReturnedIntact() + { + // A ceiling beyond int.MaxValue milliseconds must survive as a long; only the jitter + // bound saturates to int. + var big = TimeSpan.FromMilliseconds(5_000_000_000L); + var expo = Seeded(Base1s, big); + + Assert.Equal(5_000_000_000L, expo.GetUnjitteredMillisecondsForN(40)); + } + + // ---- server-directed minimum delay (SSE "retry:" field) ---- + + [Fact] + public void ServerDirectedMinDelayOverridesConfiguredMinimum() + { + var expo = Seeded(Base1s, Max30s); + + expo.SetServerDirectedMinDelay(TimeSpan.FromMilliseconds(5000)); + + Assert.Equal(5000L, expo.GetUnjitteredMillisecondsForN(0)); + Assert.Equal(10000L, expo.GetUnjitteredMillisecondsForN(1)); + } + + [Fact] + public void ServerDirectedMinDelayOverridesTemporaryMinimum() + { + var expo = Seeded(Base1s, TimeSpan.FromHours(1)); + expo.SetBounds(TimeSpan.FromMinutes(5), TimeSpan.FromHours(1)); + + expo.SetServerDirectedMinDelay(TimeSpan.FromMilliseconds(5000)); + + // The directed value wins over the temporary minimum, not just the configured one. + Assert.Equal(5000L, expo.GetUnjitteredMillisecondsForN(0)); + } + + [Fact] + public void ServerDirectedMinDelayResetsN() + { + var expo = Seeded(Base1s, Max30s); + + expo.GetNextBackOff(); + expo.GetNextBackOff(); + expo.GetNextBackOff(); + Assert.Equal(3, expo.GetBackoffN()); + + expo.SetServerDirectedMinDelay(TimeSpan.FromMilliseconds(5000)); + + // So the next attempt uses the directed value itself rather than doubling from + // wherever the progression had reached. + Assert.Equal(0, expo.GetBackoffN()); + } + + [Fact] + public void ServerDirectedMinDelayIsClampedToOneHour() + { + // The maximum is deliberately larger than the wire cap so the cap is the bound under + // test, not the ceiling. + var expo = Seeded(Base1s, TimeSpan.FromHours(24)); + + expo.SetServerDirectedMinDelay(TimeSpan.FromHours(2)); + + Assert.Equal(ExponentialBackoffWithDecorrelation.MaxServerDirectedMinDelay, + expo.GetServerDirectedMinDelay()); + Assert.Equal((long)TimeSpan.FromHours(1).TotalMilliseconds, + expo.GetUnjitteredMillisecondsForN(0)); + } + + [Fact] + public void ZeroServerDirectedMinDelayProducesZeroDelay() + { + var expo = Seeded(Base1s, Max30s); + + expo.SetServerDirectedMinDelay(TimeSpan.Zero); + + Assert.Equal(0L, expo.GetUnjitteredMillisecondsForN(0)); + Assert.Equal(0L, expo.GetUnjitteredMillisecondsForN(5)); + Assert.Equal(TimeSpan.Zero, expo.GetNextBackOff()); + } + + [Fact] + public void NegativeServerDirectedMinDelayBecomesZero() + { + var expo = Seeded(Base1s, Max30s); + + expo.SetServerDirectedMinDelay(TimeSpan.FromMilliseconds(-5)); + + Assert.Equal(TimeSpan.Zero, expo.GetServerDirectedMinDelay()); + } + + [Fact] + public void ServerDirectedMinDelayIsStillLimitedByTheMaximum() + { + var expo = Seeded(Base1s, TimeSpan.FromMilliseconds(2000)); + + expo.SetServerDirectedMinDelay(TimeSpan.FromMilliseconds(5000)); + + // The ceiling is never affected by a directed value, so it still bounds the delay. + Assert.Equal(2000L, expo.GetUnjitteredMillisecondsForN(0)); + } + + [Fact] + public void ServerDirectedMinDelayDoesNotRevertTheTemporaryCeiling() + { + // Configured ceiling 30s, temporary ceiling 1hr, then a directed value arrives. + var expo = Seeded(Base1s, Max30s); + expo.SetBounds(TimeSpan.FromMinutes(5), TimeSpan.FromHours(1)); + + expo.SetServerDirectedMinDelay(TimeSpan.FromMilliseconds(300)); + + Assert.Equal(TimeSpan.FromHours(1), expo.GetMaximumDelay()); + + // Reading at n = 0 cannot tell the two apart -- both would give 300. These levels + // can: doubling 300ms reaches the configured 30s ceiling by n = 7, so an + // implementation that reverted the regime would report 30000 here instead. + Assert.Equal(2457600L, expo.GetUnjitteredMillisecondsForN(13)); + Assert.Equal(3600000L, expo.GetUnjitteredMillisecondsForN(14)); + Assert.Equal(3600000L, expo.GetUnjitteredMillisecondsForN(62)); + } + + [Fact] + public void ServerDirectedMinDelaySurvivesSetBounds() + { + var expo = Seeded(Base1s, TimeSpan.FromHours(1)); + expo.SetServerDirectedMinDelay(TimeSpan.FromMilliseconds(5000)); + + expo.SetBounds(TimeSpan.FromMinutes(5), TimeSpan.FromHours(1)); + + Assert.Equal(TimeSpan.FromMilliseconds(5000), expo.GetServerDirectedMinDelay()); + Assert.Equal(5000L, expo.GetUnjitteredMillisecondsForN(0)); + } + + [Fact] + public void ServerDirectedMinDelaySurvivesResetBackoffN() + { + var expo = Seeded(Base1s, Max30s); + expo.SetServerDirectedMinDelay(TimeSpan.FromMilliseconds(5000)); + + expo.ResetBackoffN(); + + Assert.Equal(TimeSpan.FromMilliseconds(5000), expo.GetServerDirectedMinDelay()); + Assert.Equal(5000L, expo.GetUnjitteredMillisecondsForN(0)); + } + + [Fact] + public void LaterServerDirectedMinDelayReplacesEarlier() + { + var expo = Seeded(Base1s, Max30s); + + expo.SetServerDirectedMinDelay(TimeSpan.FromMilliseconds(5000)); + expo.SetServerDirectedMinDelay(TimeSpan.FromMilliseconds(7000)); + + Assert.Equal(TimeSpan.FromMilliseconds(7000), expo.GetServerDirectedMinDelay()); + Assert.Equal(7000L, expo.GetUnjitteredMillisecondsForN(0)); + } + + [Fact] + public void NoServerDirectedMinDelayByDefault() + { + var expo = Seeded(Base1s, Max30s); + + Assert.Null(expo.GetServerDirectedMinDelay()); + } + + // ---- jitter ---- + + [Fact] + public void JitteredDelayStaysBetweenHalfAndFullOfUnjitteredValue() + { + var expo = Seeded(Base1s, Max30s); + + for (int level = 0; level < 8; level++) + { + long unjittered = expo.GetUnjitteredMillisecondsForN(level); + var delay = expo.GetNextBackOff(); + + Assert.InRange(delay.TotalMilliseconds, unjittered / 2, unjittered); + } + } + + [Fact] + public void JitterAppliesToAMaximumAboveIntRangeWithoutThrowing() + { + // The jitter bound saturates at int.MaxValue; the delay itself must stay a long. + var expo = Seeded(TimeSpan.FromMilliseconds(4_000_000_000L), + TimeSpan.FromMilliseconds(8_000_000_000L)); + + var delay = expo.GetNextBackOff(); + + Assert.InRange(delay, TimeSpan.FromMilliseconds(2_000_000_000L), + TimeSpan.FromMilliseconds(4_000_000_000L)); + } + + [Fact] + public void ZeroDelayProducesZeroWithoutThrowing() + { + var expo = Seeded(TimeSpan.Zero, TimeSpan.Zero); + + Assert.Equal(TimeSpan.Zero, expo.GetNextBackOff()); + Assert.Equal(TimeSpan.Zero, expo.GetNextBackOff()); + } + + [Fact] + public void TinyMinimumDelayProducesNoJitter() + { + // With a 1ms base the integer arithmetic leaves no room for a jitter component. + var expo = Seeded(TimeSpan.FromMilliseconds(1), TimeSpan.FromMilliseconds(1)); + + Assert.Equal(TimeSpan.Zero, expo.GetNextBackOff()); + } + + [Fact] + public void SeededRandomProducesDeterministicDelays() + { + var a = Seeded(Base1s, Max30s); + var b = Seeded(Base1s, Max30s); + + for (int i = 0; i < 5; i++) + { + Assert.Equal(a.GetNextBackOff(), b.GetNextBackOff()); + } + } + + [Fact] + public void DelaysAreWholeMilliseconds() + { + // Nothing in the delay path passes through a floating-point value, so every result + // lands on a millisecond boundary. + var expo = Seeded(Base1s, Max30s); + + for (int i = 0; i < 10; i++) + { + var delay = expo.GetNextBackOff(); + Assert.Equal(0, delay.Ticks % TimeSpan.TicksPerMillisecond); + } + } + + // ---- n ---- + + [Fact] + public void SetBoundsResetsBackoffN() + { + var expo = Seeded(Base1s, Max30s); + + expo.GetNextBackOff(); + expo.GetNextBackOff(); + expo.GetNextBackOff(); + Assert.Equal(3, expo.GetBackoffN()); + + expo.SetBounds(TimeSpan.FromMinutes(5), TimeSpan.FromHours(1)); + + Assert.Equal(0, expo.GetBackoffN()); + } + + [Fact] + public void ResetBackoffNReturnsToZero() + { + var expo = Seeded(Base1s, Max30s); + + expo.GetNextBackOff(); + expo.GetNextBackOff(); + + expo.ResetBackoffN(); + + Assert.Equal(0, expo.GetBackoffN()); + } + + [Fact] + public void SetBoundsWithSameValuesIsNoOp() + { + var expo = Seeded(Base1s, Max30s); + + expo.GetNextBackOff(); + expo.GetNextBackOff(); + Assert.Equal(2, expo.GetBackoffN()); + + expo.SetBounds(Base1s, Max30s); + + Assert.Equal(2, expo.GetBackoffN()); + } + + [Fact] + public void SetBoundsRepeatedlyWithSameValuesDoesNotPinTheDelay() + { + var expo = Seeded(TimeSpan.FromMinutes(5), TimeSpan.FromHours(1)); + + // A caller reapplying identical bounds on every failure must still see the backoff + // level advance. + for (int i = 0; i < 4; i++) + { + expo.SetBounds(TimeSpan.FromMinutes(5), TimeSpan.FromHours(1)); + expo.GetNextBackOff(); + } + + Assert.Equal(4, expo.GetBackoffN()); + } + + [Fact] + public void SetBoundsAppliesNewValues() + { + var expo = Seeded(Base1s, Max30s); + + expo.SetBounds(TimeSpan.FromMinutes(5), TimeSpan.FromHours(1)); + + Assert.Equal(TimeSpan.FromMinutes(5), expo.GetMinimumDelay()); + Assert.Equal(TimeSpan.FromHours(1), expo.GetMaximumDelay()); + Assert.Equal((long)TimeSpan.FromMinutes(5).TotalMilliseconds, + expo.GetUnjitteredMillisecondsForN(0)); + } + + [Fact] + public void NegativeBoundsClampToZero() + { + var expo = Seeded(TimeSpan.FromMilliseconds(-1000), TimeSpan.FromMilliseconds(-1)); + + Assert.Equal(TimeSpan.Zero, expo.GetMinimumDelay()); + Assert.Equal(TimeSpan.Zero, expo.GetMaximumDelay()); + + expo.SetBounds(TimeSpan.FromMilliseconds(-5), TimeSpan.FromMilliseconds(-5)); + Assert.Equal(TimeSpan.Zero, expo.GetMinimumDelay()); + Assert.Equal(TimeSpan.Zero, expo.GetMaximumDelay()); + } + + [Fact] + public void SetBoundsToNegativeIsNoOpWhenAlreadyZero() + { + var expo = Seeded(TimeSpan.Zero, TimeSpan.Zero); + + expo.GetNextBackOff(); + Assert.Equal(1, expo.GetBackoffN()); + + // Clamping happens before the comparison, so this must not reset the backoff level. + expo.SetBounds(TimeSpan.FromMilliseconds(-10), TimeSpan.FromMilliseconds(-20)); + Assert.Equal(1, expo.GetBackoffN()); + } } } diff --git a/test/LaunchDarkly.EventSource.Tests/ServerDirectedRetryDelayTest.cs b/test/LaunchDarkly.EventSource.Tests/ServerDirectedRetryDelayTest.cs new file mode 100644 index 0000000..083f9b6 --- /dev/null +++ b/test/LaunchDarkly.EventSource.Tests/ServerDirectedRetryDelayTest.cs @@ -0,0 +1,226 @@ +using System; +using System.Threading.Tasks; +using LaunchDarkly.TestHelpers.HttpTest; +using Xunit; +using Xunit.Abstractions; + +using static LaunchDarkly.EventSource.Tests.TestHelpers; + +namespace LaunchDarkly.EventSource.Tests +{ + /// + /// Covers the SSE retry: field: a server-directed reconnection time that overrides the + /// minimum delay used for subsequent backoffs. + /// + public class ServerDirectedRetryDelayTest : BaseTest + { + private static readonly TimeSpan ConfiguredInitial = TimeSpan.FromMilliseconds(20); + private static readonly TimeSpan ConfiguredMax = TimeSpan.FromHours(24); + private static readonly MessageEvent Marker = new MessageEvent("put", "hello", _uri); + + public ServerDirectedRetryDelayTest(ITestOutputHelper testOutput) : base(testOutput) { } + + // Emits a raw retry: line, then a normal event. Once the event arrives we know the retry + // line has definitely been processed -- the same sequencing Java's tests use. + private static Handler StreamWithRetryLine(string value) => + StartStream() + .Then(Handlers.WriteChunkString("retry: " + value + "\n")) + .Then(WriteEvent(Marker)) + .Then(LeaveStreamOpen()); + + private void WithRetryLine(string value, Action assert, + Action modConfig = null) + { + WithServerAndEventSource(StreamWithRetryLine(value), + c => + { + c.InitialRetryDelay(ConfiguredInitial).MaxRetryDelay(ConfiguredMax) + .BackoffResetThreshold(TimeSpan.FromSeconds(30)); + modConfig?.Invoke(c); + }, + (server, es) => + { + var sink = new EventSink(es, _testLogging); + _ = Task.Run(es.StartAsync); + + sink.ExpectActions( + EventSink.OpenedAction(), + EventSink.MessageReceivedAction(Marker)); + + assert(es); + }); + } + + [Fact] + public void RetryFieldSetsTheMinDelay() + { + WithRetryLine("300", es => + { + Assert.Equal(TimeSpan.FromMilliseconds(300), + es.BackOff.GetServerDirectedMinDelay()); + Assert.Equal(300L, es.BackOff.GetUnjitteredMillisecondsForN(0)); + }); + } + + [Fact] + public void RetryFieldOverridesTheConfiguredInitialDelay() + { + WithRetryLine("300", es => + Assert.NotEqual(ConfiguredInitial, es.BackOff.GetServerDirectedMinDelay())); + } + + [Fact] + public void RetryFieldOverridesTemporaryBounds() + { + WithRetryLine("300", es => + { + es.SetTemporaryRetryDelayBounds(TimeSpan.FromMinutes(5), TimeSpan.FromHours(1)); + + // The directed value outranks the temporary minimum, matching the reference + // implementations, which apply a wire hint to every registered regime. + Assert.Equal(300L, es.BackOff.GetUnjitteredMillisecondsForN(0)); + }); + } + + [Fact] + public void RetryFieldDoesNotRevertAnActiveTemporaryCeiling() + { + WithServerAndEventSource(StreamWithRetryLine("300"), + // A 30s configured ceiling, distinct from the temporary one, so the assertion + // can tell which is in effect. The reset threshold is long enough that the + // healthy-op path cannot clear the temporary bounds mid-test. + c => c.InitialRetryDelay(ConfiguredInitial) + .MaxRetryDelay(TimeSpan.FromSeconds(30)) + .BackoffResetThreshold(TimeSpan.FromSeconds(30)), + (server, es) => + { + var sink = new EventSink(es, _testLogging); + + // The extended regime is already active when the retry: line arrives. + es.SetTemporaryRetryDelayBounds(TimeSpan.FromMinutes(5), TimeSpan.FromHours(1)); + + _ = Task.Run(es.StartAsync); + sink.ExpectActions( + EventSink.OpenedAction(), + EventSink.MessageReceivedAction(Marker)); + + Assert.Equal(TimeSpan.FromMilliseconds(300), + es.BackOff.GetServerDirectedMinDelay()); + + // The directed value replaces the minimum only. The ceiling stays the + // temporary one, so reconnection is still bounded by 1hr rather than + // dropping back to the configured 30s. + Assert.Equal(TimeSpan.FromHours(1), es.BackOff.GetMaximumDelay()); + Assert.Equal(3600000L, es.BackOff.GetUnjitteredMillisecondsForN(14)); + }); + } + + [Fact] + public void RetryFieldAboveOneHourIsClampedToOneHour() + { + // ConfiguredMax is 24 hours so the wire cap, not the ceiling, is the bound under test. + WithRetryLine("7200000", es => + Assert.Equal(ExponentialBackoffWithDecorrelation.MaxServerDirectedMinDelay, + es.BackOff.GetServerDirectedMinDelay())); + } + + [Fact] + public void RetryFieldOfZeroDisablesBackoff() + { + WithRetryLine("0", es => + { + Assert.Equal(TimeSpan.Zero, es.BackOff.GetServerDirectedMinDelay()); + Assert.Equal(0L, es.BackOff.GetUnjitteredMillisecondsForN(0)); + Assert.Equal(TimeSpan.Zero, es.BackOff.GetNextBackOff()); + }); + } + + [Theory] + [InlineData("7000L")] + [InlineData("3.5")] + [InlineData("1e3")] + [InlineData("1,000")] + [InlineData("")] + [InlineData("abc")] + public void UnparseableRetryFieldIsIgnored(string value) + { + WithRetryLine(value, es => + Assert.Null(es.BackOff.GetServerDirectedMinDelay())); + } + + [Fact] + public void NegativeRetryFieldClampsToZeroAndDisablesBackoff() + { + // The RETRY specification imposes no floor on server-directed values, so a negative + // one clamps to zero and zero means reconnect without waiting. + WithRetryLine("-5", es => + { + Assert.Equal(TimeSpan.Zero, es.BackOff.GetServerDirectedMinDelay()); + Assert.Equal(TimeSpan.Zero, es.BackOff.GetNextBackOff()); + }); + } + + [Fact] + public void RetryFieldWithLeadingSignIsAccepted() + { + WithRetryLine("+5", es => + Assert.Equal(TimeSpan.FromMilliseconds(5), + es.BackOff.GetServerDirectedMinDelay())); + } + + [Fact] + public void RetryFieldSurvivesSustainedHealthyConnection() + { + // A zero reset threshold makes the healthy-op reset fire deterministically on the + // reconnect. The directed value must survive it. + // The first stream must end rather than stay open, so that a reconnect happens and + // the healthy-op reset runs. + var handler = Handlers.Sequential( + StartStream() + .Then(Handlers.WriteChunkString("retry: 300\n")) + .Then(WriteEvent(Marker)), + StartStream().Then(WriteEvent(Marker)).Then(LeaveStreamOpen()) + ); + + WithServerAndEventSource(handler, + c => c.InitialRetryDelay(ConfiguredInitial).MaxRetryDelay(ConfiguredMax) + .BackoffResetThreshold(TimeSpan.Zero), + (server, es) => + { + _ = new EventSink(es, _testLogging); + _ = Task.Run(es.StartAsync); + + server.Recorder.RequireRequest(); + server.Recorder.RequireRequest(); + + Assert.Equal(TimeSpan.FromMilliseconds(300), + es.BackOff.GetServerDirectedMinDelay()); + }); + } + + [Fact] + public void LaterRetryFieldReplacesEarlier() + { + var handler = StartStream() + .Then(Handlers.WriteChunkString("retry: 300\n")) + .Then(Handlers.WriteChunkString("retry: 700\n")) + .Then(WriteEvent(Marker)) + .Then(LeaveStreamOpen()); + + WithServerAndEventSource(handler, + c => c.InitialRetryDelay(ConfiguredInitial).MaxRetryDelay(ConfiguredMax), + (server, es) => + { + var sink = new EventSink(es, _testLogging); + _ = Task.Run(es.StartAsync); + + sink.ExpectActions( + EventSink.OpenedAction(), + EventSink.MessageReceivedAction(Marker)); + + Assert.Equal(TimeSpan.FromMilliseconds(700), + es.BackOff.GetServerDirectedMinDelay()); + }); + } + } +} diff --git a/test/LaunchDarkly.EventSource.Tests/TemporaryRetryDelayBoundsTest.cs b/test/LaunchDarkly.EventSource.Tests/TemporaryRetryDelayBoundsTest.cs new file mode 100644 index 0000000..f4faccb --- /dev/null +++ b/test/LaunchDarkly.EventSource.Tests/TemporaryRetryDelayBoundsTest.cs @@ -0,0 +1,337 @@ +using System; +using System.Net; +using System.Threading.Tasks; +using LaunchDarkly.TestHelpers.HttpTest; +using Xunit; +using Xunit.Abstractions; + +using static LaunchDarkly.EventSource.Tests.TestHelpers; + +namespace LaunchDarkly.EventSource.Tests +{ + public class TemporaryRetryDelayBoundsTest : BaseTest + { + private static readonly TimeSpan ConfiguredInitial = TimeSpan.FromMilliseconds(20); + private static readonly TimeSpan ConfiguredMax = TimeSpan.FromMilliseconds(200); + private static readonly TimeSpan TempInitial = TimeSpan.FromMilliseconds(60); + private static readonly TimeSpan TempMax = TimeSpan.FromMilliseconds(500); + + // Long enough that no test connection can reach it, so temporary bounds are never + // auto-cleared in tests that don't want them to be. + private static readonly TimeSpan NeverResets = TimeSpan.FromSeconds(30); + + public TemporaryRetryDelayBoundsTest(ITestOutputHelper testOutput) : base(testOutput) { } + + private EventSource MakeConfigured(Action extra = null) => + MakeEventSource(_uri, c => + { + c.InitialRetryDelay(ConfiguredInitial).MaxRetryDelay(ConfiguredMax); + extra?.Invoke(c); + }); + + // ---- bounds installation and reversion ---- + + [Fact] + public void FreshEventSourceUsesConfiguredBounds() + { + using (var es = MakeConfigured()) + { + Assert.Equal(ConfiguredInitial, es.BackOff.GetMinimumDelay()); + Assert.Equal(ConfiguredMax, es.BackOff.GetMaximumDelay()); + } + } + + [Fact] + public void SetTemporaryBoundsChangesTheActiveBounds() + { + using (var es = MakeConfigured()) + { + es.SetTemporaryRetryDelayBounds(TempInitial, TempMax); + + Assert.Equal(TempInitial, es.BackOff.GetMinimumDelay()); + Assert.Equal(TempMax, es.BackOff.GetMaximumDelay()); + } + } + + [Fact] + public void ClearTemporaryBoundsRestoresConfiguredBounds() + { + using (var es = MakeConfigured()) + { + es.SetTemporaryRetryDelayBounds(TempInitial, TempMax); + es.ClearTemporaryRetryDelayBounds(); + + Assert.Equal(ConfiguredInitial, es.BackOff.GetMinimumDelay()); + Assert.Equal(ConfiguredMax, es.BackOff.GetMaximumDelay()); + } + } + + [Fact] + public void ClearTemporaryBoundsWithNoneSetIsNoOp() + { + using (var es = MakeConfigured()) + { + es.BackOff.GetNextBackOff(); + es.BackOff.GetNextBackOff(); + + es.ClearTemporaryRetryDelayBounds(); + + // The configured bounds were already in effect, so nothing changed -- in particular + // the backoff level was not reset. + Assert.Equal(2, es.BackOff.GetBackoffN()); + Assert.Equal(ConfiguredInitial, es.BackOff.GetMinimumDelay()); + } + } + + // ---- backoff level vs. reconnect attempt count ---- + + [Fact] + public void SetTemporaryBoundsResetsBackoffN() + { + using (var es = MakeConfigured()) + { + es.BackOff.GetNextBackOff(); + es.BackOff.GetNextBackOff(); + es.BackOff.GetNextBackOff(); + + es.SetTemporaryRetryDelayBounds(TempInitial, TempMax); + + Assert.Equal(0, es.BackOff.GetBackoffN()); + } + } + + [Fact] + public void SettingTheSameTemporaryBoundsAgainDoesNotResetTheBackoffN() + { + using (var es = MakeConfigured()) + { + es.SetTemporaryRetryDelayBounds(TempInitial, TempMax); + es.BackOff.GetNextBackOff(); + es.BackOff.GetNextBackOff(); + + es.SetTemporaryRetryDelayBounds(TempInitial, TempMax); + + Assert.Equal(2, es.BackOff.GetBackoffN()); + } + } + + [Fact] + public void RepeatedlySettingTemporaryBoundsDoesNotPinTheDelay() + { + using (var es = MakeConfigured()) + { + // A consumer that reapplies the same bounds on every error must still back off. + for (int i = 0; i < 5; i++) + { + es.SetTemporaryRetryDelayBounds(TempInitial, TempMax); + es.BackOff.GetNextBackOff(); + } + + Assert.Equal(5, es.BackOff.GetBackoffN()); + } + } + + [Fact] + public void ReEnteringTemporaryBoundsAfterClearResetsTheBackoffN() + { + using (var es = MakeConfigured()) + { + es.SetTemporaryRetryDelayBounds(TempInitial, TempMax); + es.BackOff.GetNextBackOff(); + es.BackOff.GetNextBackOff(); + es.BackOff.GetNextBackOff(); + Assert.Equal(3, es.BackOff.GetBackoffN()); + + es.ClearTemporaryRetryDelayBounds(); + es.SetTemporaryRetryDelayBounds(TempInitial, TempMax); + + // The progression restarts rather than resuming where it left off, so a connection + // that recovered and then failed again is probed sooner. + Assert.Equal(0, es.BackOff.GetBackoffN()); + } + } + + // ---- argument handling ---- + + [Fact] + public void TemporaryBoundsWithInitialAboveMaxAreLimitedToMax() + { + using (var es = MakeConfigured()) + { + es.SetTemporaryRetryDelayBounds(TimeSpan.FromSeconds(10), TimeSpan.FromSeconds(1)); + + Assert.Equal(1000L, es.BackOff.GetUnjitteredMillisecondsForN(0)); + Assert.Equal(1000L, es.BackOff.GetUnjitteredMillisecondsForN(4)); + } + } + + [Fact] + public void NegativeTemporaryBoundsClampToZeroWithoutThrowing() + { + using (var es = MakeConfigured()) + { + es.SetTemporaryRetryDelayBounds(TimeSpan.FromMilliseconds(-10), + TimeSpan.FromMilliseconds(-20)); + + Assert.Equal(TimeSpan.Zero, es.BackOff.GetMinimumDelay()); + Assert.Equal(TimeSpan.Zero, es.BackOff.GetMaximumDelay()); + } + } + + [Fact] + public void AbsurdTemporaryBoundsCannotBreakTheReconnectLoop() + { + using (var es = MakeConfigured()) + { + es.SetTemporaryRetryDelayBounds(TimeSpan.FromMilliseconds(1000), TimeSpan.MaxValue); + + // No caller-supplied bound may make a later delay computation throw; that would + // escape StartAsync and stop the stream for good. + for (int i = 0; i < 64; i++) + { + es.BackOff.GetNextBackOff(); + } + } + } + + [Fact] + public void SetTemporaryBoundsIsSafeFromAnotherThread() + { + using (var es = MakeConfigured()) + { + const int iterations = 500; + var writer = Task.Run(() => + { + for (int i = 0; i < iterations; i++) + { + es.SetTemporaryRetryDelayBounds( + TimeSpan.FromMilliseconds(10 + (i % 2)), + TimeSpan.FromMilliseconds(100 + (i % 2))); + } + }); + + int computed = 0; + for (int i = 0; i < iterations; i++) + { + es.BackOff.GetNextBackOff(); + computed++; + } + writer.Wait(TimeSpan.FromSeconds(10)); + + Assert.True(writer.IsCompleted); + Assert.Equal(iterations, computed); + // Bounds must end up as one of the two pairs the writer set, never a torn mix. + Assert.Contains(es.BackOff.GetMinimumDelay().TotalMilliseconds, new double[] { 10, 11 }); + Assert.Contains(es.BackOff.GetMaximumDelay().TotalMilliseconds, new double[] { 100, 101 }); + } + } + + // ---- wiring through a live stream ---- + + [Fact] + public void BoundsSetFromErrorHandlerTakeEffect() + { + var handler = Handlers.Sequential( + Handlers.Status((int)HttpStatusCode.Unauthorized), + StartStream().Then(LeaveStreamOpen()) + ); + + WithServerAndEventSource(handler, + c => c.InitialRetryDelay(ConfiguredInitial).MaxRetryDelay(ConfiguredMax) + .BackoffResetThreshold(NeverResets), + (server, es) => + { + es.Error += (_, e) => es.SetTemporaryRetryDelayBounds(TempInitial, TempMax); + + _ = Task.Run(es.StartAsync); + + server.Recorder.RequireRequest(); + server.Recorder.RequireRequest(); // the reconnect, which used the new bounds + + Assert.Equal(TempInitial, es.BackOff.GetMinimumDelay()); + Assert.Equal(TempMax, es.BackOff.GetMaximumDelay()); + }); + } + + [Fact] + public void SustainedHealthyConnectionClearsTemporaryBounds() + { + var handler = Handlers.Sequential( + StartStream().Then(WriteEvent(new MessageEvent("put", "hello", _uri))), + StartStream().Then(LeaveStreamOpen()) + ); + + WithServerAndEventSource(handler, + // A zero threshold means any connection that opened counts as healthy, which makes + // the reset deterministic rather than timing-dependent. + c => c.InitialRetryDelay(ConfiguredInitial).MaxRetryDelay(ConfiguredMax) + .BackoffResetThreshold(TimeSpan.Zero), + (server, es) => + { + es.SetTemporaryRetryDelayBounds(TempInitial, TempMax); + + _ = Task.Run(es.StartAsync); + + server.Recorder.RequireRequest(); + server.Recorder.RequireRequest(); + + Assert.Equal(ConfiguredInitial, es.BackOff.GetMinimumDelay()); + Assert.Equal(ConfiguredMax, es.BackOff.GetMaximumDelay()); + }); + } + + [Fact] + public void ConnectionShorterThanThresholdDoesNotClearTemporaryBounds() + { + var handler = Handlers.Sequential( + StartStream().Then(WriteEvent(new MessageEvent("put", "hello", _uri))), + StartStream().Then(LeaveStreamOpen()) + ); + + WithServerAndEventSource(handler, + c => c.InitialRetryDelay(ConfiguredInitial).MaxRetryDelay(ConfiguredMax) + .BackoffResetThreshold(NeverResets), + (server, es) => + { + es.SetTemporaryRetryDelayBounds(TempInitial, TempMax); + + _ = Task.Run(es.StartAsync); + + server.Recorder.RequireRequest(); + server.Recorder.RequireRequest(); + + Assert.Equal(TempInitial, es.BackOff.GetMinimumDelay()); + Assert.Equal(TempMax, es.BackOff.GetMaximumDelay()); + }); + } + + [Fact] + public void RestartDoesNotRevertTemporaryBounds() + { + // The chunk write is what flushes the response headers, so that Opened fires. + var handler = StartStream().Then(Handlers.WriteChunkString(":hi\n")) + .Then(LeaveStreamOpen()); + + WithServerAndEventSource(handler, + c => c.InitialRetryDelay(ConfiguredInitial).MaxRetryDelay(ConfiguredMax) + .BackoffResetThreshold(NeverResets), + (server, es) => + { + var sink = new EventSink(es, _testLogging); + + _ = Task.Run(es.StartAsync); + sink.ExpectActions(EventSink.OpenedAction()); + + es.SetTemporaryRetryDelayBounds(TempInitial, TempMax); + es.Restart(true); + + server.Recorder.RequireRequest(); + + // Restart resets the backoff level, but within whatever bounds are active -- + // it does not discard temporary bounds. + Assert.Equal(TempInitial, es.BackOff.GetMinimumDelay()); + Assert.Equal(TempMax, es.BackOff.GetMaximumDelay()); + }); + } + } +} From b786b7d61c7d84f2b78ecab625e9454b8891c3a5 Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Mon, 31 Aug 2026 10:11:58 -0400 Subject: [PATCH 2/5] fix: interrupt a pending backoff wait on shutdown Close() could not end a backoff wait already in progress: Task.Delay was given no cancellation token, and the per-request token is created after the wait, so CancelCurrentRequest had nothing to act on. The reconnect loop stayed alive until the delay elapsed, holding the EventSource and its timer. That was previously bounded by MaxRetryDelay, 30 seconds by default. Temporary bounds and server-directed values now make an hour reachable, so the wait takes a shutdown token that Close cancels. RETRY specification Requirement 1.10.1. SDK-2791 --- src/LaunchDarkly.EventSource/EventSource.cs | 11 ++++++- .../EventSourceReconnectingTest.cs | 33 +++++++++++++++++++ 2 files changed, 43 insertions(+), 1 deletion(-) diff --git a/src/LaunchDarkly.EventSource/EventSource.cs b/src/LaunchDarkly.EventSource/EventSource.cs index 2491ea3..6a58988 100644 --- a/src/LaunchDarkly.EventSource/EventSource.cs +++ b/src/LaunchDarkly.EventSource/EventSource.cs @@ -28,6 +28,7 @@ public class EventSource : IEventSource, IDisposable private readonly TimeSpan _configuredInitialRetryDelay; private readonly TimeSpan _configuredMaxRetryDelay; private CancellationTokenSource _currentRequestToken; + private readonly CancellationTokenSource _shutdownTokenSource = new CancellationTokenSource(); private DateTime? _lastSuccessfulConnectionTime; private ReadyState _readyState; @@ -214,7 +215,14 @@ private async Task MaybeWaitWithBackOff() { { _logger.Info("Waiting {0} milliseconds before reconnecting...", sleepTime.TotalMilliseconds); BackOffDelay = sleepTime; - await Task.Delay(sleepTime); + try + { + await Task.Delay(sleepTime, _shutdownTokenSource.Token); + } + catch (OperationCanceledException) + { + // Cancellation happened during the wait, likely intentional close + } } } @@ -257,6 +265,7 @@ public void Close() Close(ReadyState.Shutdown); } CancelCurrentRequest(); + _shutdownTokenSource.Cancel(); // do not dispose httpClient if it is user provided if (_configuration.HttpClient == null) diff --git a/test/LaunchDarkly.EventSource.Tests/EventSourceReconnectingTest.cs b/test/LaunchDarkly.EventSource.Tests/EventSourceReconnectingTest.cs index 263e9a8..996bbdd 100644 --- a/test/LaunchDarkly.EventSource.Tests/EventSourceReconnectingTest.cs +++ b/test/LaunchDarkly.EventSource.Tests/EventSourceReconnectingTest.cs @@ -108,6 +108,39 @@ public void RetryDelayDurationsShouldIncrease() AssertBackoffsAlwaysIncrease(backoffs, nAttempts); } + [Fact] + public async Task CloseInterruptsAPendingBackoffWait() + { + // The first attempt fails, so the loop enters a backoff wait long enough that + // running it to completion would hang the test rather than merely slow it. + var handler = Handlers.Sequential( + Handlers.Status((int)HttpStatusCode.InternalServerError), + StartStream().Then(LeaveStreamOpen()) + ); + + using (var server = HttpServer.Start(handler)) + { + using (var es = MakeEventSource(server.Uri, + c => c.InitialRetryDelay(TimeSpan.FromMinutes(10)))) + { + var eventSink = new EventSink(es, _testLogging); + var streamTask = Task.Run(es.StartAsync); + + // Waiting for the Closed action means the failed attempt has been reported, + // so the loop is at or about to reach the wait. + eventSink.ExpectAction(); + eventSink.ExpectActions(EventSink.ClosedAction()); + + es.Close(); + + var finished = await Task.WhenAny(streamTask, + Task.Delay(TimeSpan.FromSeconds(5))); + + Assert.Same(streamTask, finished); + } + } + } + [Fact] public async Task NoReconnectAttemptIsMadeIfErrorHandlerClosesEventSource() { From 6c7718254bc9a99d2b01985e86f4a1297ff7549b Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Mon, 31 Aug 2026 10:30:54 -0400 Subject: [PATCH 3/5] test: avoid a blocking task wait in the concurrency test xUnit1031 flags Task.Wait inside a test method. Replacing it with WhenAny plus an await keeps the deadlock guard and additionally surfaces an exception thrown on the writer thread, which Wait followed by IsCompleted did not. --- .../TemporaryRetryDelayBoundsTest.cs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/test/LaunchDarkly.EventSource.Tests/TemporaryRetryDelayBoundsTest.cs b/test/LaunchDarkly.EventSource.Tests/TemporaryRetryDelayBoundsTest.cs index f4faccb..ed8e161 100644 --- a/test/LaunchDarkly.EventSource.Tests/TemporaryRetryDelayBoundsTest.cs +++ b/test/LaunchDarkly.EventSource.Tests/TemporaryRetryDelayBoundsTest.cs @@ -195,7 +195,7 @@ public void AbsurdTemporaryBoundsCannotBreakTheReconnectLoop() } [Fact] - public void SetTemporaryBoundsIsSafeFromAnotherThread() + public async Task SetTemporaryBoundsIsSafeFromAnotherThread() { using (var es = MakeConfigured()) { @@ -216,9 +216,11 @@ public void SetTemporaryBoundsIsSafeFromAnotherThread() es.BackOff.GetNextBackOff(); computed++; } - writer.Wait(TimeSpan.FromSeconds(10)); + // Bounded so that a genuine deadlock fails the test rather than hanging it. + var finished = await Task.WhenAny(writer, Task.Delay(TimeSpan.FromSeconds(10))); + Assert.Same(writer, finished); + await writer; // surfaces any exception thrown on the writer thread - Assert.True(writer.IsCompleted); Assert.Equal(iterations, computed); // Bounds must end up as one of the two pairs the writer set, never a torn mix. Assert.Contains(es.BackOff.GetMinimumDelay().TotalMilliseconds, new double[] { 10, 11 }); From 7a39b04940b6454811386e779f0ef17058023cd2 Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Tue, 1 Sep 2026 11:41:24 -0400 Subject: [PATCH 4/5] fix: address first-round review findings Four behavior fixes, plus documentation corrections and test coverage. - Close() from the Raw state did not transition ReadyState, so a StartAsync running afterwards saw Raw and began connecting: an unbounded reconnect loop. Every state except Shutdown now transitions, so ReadyState reports Shutdown and Closed(Shutdown) fires where that path was previously silent. This was a regression introduced by b786b7d. - A computed delay above Task.Delay's platform ceiling threw out of StartAsync, stopping the stream permanently. Now clamped at int.MaxValue milliseconds, the lower of the two target frameworks' ceilings. - A large retry: value threw OverflowException from TimeSpan.FromMilliseconds before the one-hour clamp could apply, tearing the connection down on every read of the line and applying nothing at all. Now clamped in the integer domain at the wire boundary, which is also where Java and Go clamp. The overflow is symmetric, so both directions are covered. - The healthy-operation reset window was measured with DateTime.Now, a local wall clock, making a daylight-saving transition a deterministic twice-yearly trigger and letting an NTP step fire the reset early or suppress it entirely. Now measured with a monotonic Stopwatch. Documentation: the public docs claimed ClearTemporaryRetryDelayBounds and the healthy-operation restore bring back the configured InitialRetryDelay, which does not hold once a retry: value has been received, since the effective minimum resolves to the server-directed value first. Corrected in six places across IEventSource and ConfigurationBuilder. Also restored the locking precondition on the delay computation, which had been lost in a rename. Tests: an exact BigInteger reference for the delay formula, over curated bound pairs and a million randomized checks, pinning the shift-based arithmetic to a reference that cannot overflow; and an end-to-end test that a retry: value changes the actual wait rather than only the backoff's internal state. --- .../ConfigurationBuilder.cs | 9 +- src/LaunchDarkly.EventSource/Constants.cs | 5 + src/LaunchDarkly.EventSource/EventSource.cs | 27 +++-- .../ExponentialBackoffWithDecorrelation.cs | 18 +-- src/LaunchDarkly.EventSource/IEventSource.cs | 25 +++- .../EventSourceReconnectingTest.cs | 65 ++++++++++ ...ExponentialBackoffWithDecorrelationTest.cs | 112 +++++++++++++++--- .../ServerDirectedRetryDelayTest.cs | 87 +++++++++++++- .../TemporaryRetryDelayBoundsTest.cs | 34 ++++-- 9 files changed, 338 insertions(+), 44 deletions(-) diff --git a/src/LaunchDarkly.EventSource/ConfigurationBuilder.cs b/src/LaunchDarkly.EventSource/ConfigurationBuilder.cs index a91d749..f87e648 100644 --- a/src/LaunchDarkly.EventSource/ConfigurationBuilder.cs +++ b/src/LaunchDarkly.EventSource/ConfigurationBuilder.cs @@ -97,6 +97,11 @@ public ConfigurationBuilder HttpRequestModifier(Action httpR /// a backoff algorithm. /// /// + /// A server-directed reconnection time received via the SSE retry: field supersedes + /// this value, as do bounds installed by + /// . + /// + /// /// The default value is . Negative values /// are changed to zero. /// @@ -150,7 +155,9 @@ public ConfigurationBuilder MaxRetryDelay(TimeSpan maxRetryDelay) /// /// A connection that stays open for at least this long also discards any bounds set by /// , restoring - /// and . + /// and . A + /// server-directed reconnection time received via the SSE retry: field is not + /// discarded. /// /// /// The default value is . Negative diff --git a/src/LaunchDarkly.EventSource/Constants.cs b/src/LaunchDarkly.EventSource/Constants.cs index fb60edf..da18b4d 100644 --- a/src/LaunchDarkly.EventSource/Constants.cs +++ b/src/LaunchDarkly.EventSource/Constants.cs @@ -25,6 +25,11 @@ internal static class Constants /// internal static string RetryField = "retry"; + /// + /// The largest reconnection time honored from a Server Sent Event retry field. + /// + internal const long MaxServerDirectedRetryDelayMillis = 3_600_000; + /// /// The identifier field name in a Server Sent Event. /// diff --git a/src/LaunchDarkly.EventSource/EventSource.cs b/src/LaunchDarkly.EventSource/EventSource.cs index 6a58988..4169ceb 100644 --- a/src/LaunchDarkly.EventSource/EventSource.cs +++ b/src/LaunchDarkly.EventSource/EventSource.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Diagnostics; using System.IO; using System.Net.Http; using System.Threading; @@ -29,7 +30,8 @@ public class EventSource : IEventSource, IDisposable private readonly TimeSpan _configuredMaxRetryDelay; private CancellationTokenSource _currentRequestToken; private readonly CancellationTokenSource _shutdownTokenSource = new CancellationTokenSource(); - private DateTime? _lastSuccessfulConnectionTime; + private static readonly TimeSpan MaxSleepTime = TimeSpan.FromMilliseconds(int.MaxValue); + private readonly Stopwatch _connectionTimer = new Stopwatch(); private ReadyState _readyState; #endregion @@ -128,9 +130,9 @@ public async Task StartAsync() { if (!firstTime) { - if (_lastSuccessfulConnectionTime.HasValue) + if (_connectionTimer.IsRunning) { - if (DateTime.Now.Subtract(_lastSuccessfulConnectionTime.Value) >= _configuration.BackoffResetThreshold) + if (_connectionTimer.Elapsed >= _configuration.BackoffResetThreshold) { // Sustained healthy operation reverts any temporary bounds and resets // n. Reverting takes precedence over bounds that were @@ -138,7 +140,7 @@ public async Task StartAsync() ClearTemporaryRetryDelayBounds(); _backOff.ResetBackoffN(); } - _lastSuccessfulConnectionTime = null; + _connectionTimer.Reset(); } await MaybeWaitWithBackOff(); } @@ -211,6 +213,13 @@ public async Task StartAsync() private async Task MaybeWaitWithBackOff() { TimeSpan sleepTime = _backOff.GetNextBackOff(); + if (sleepTime > MaxSleepTime) + { + // Task.Delay throws for anything above the platform timer ceiling, and this method + // runs outside the reconnect loop's exception handling, so an unclamped value + // would fault StartAsync and stop the stream permanently. + sleepTime = MaxSleepTime; + } if (sleepTime > TimeSpan.Zero) { _logger.Info("Waiting {0} milliseconds before reconnecting...", sleepTime.TotalMilliseconds); @@ -222,6 +231,7 @@ private async Task MaybeWaitWithBackOff() { catch (OperationCanceledException) { // Cancellation happened during the wait, likely intentional close + _logger.Debug("Backoff wait interrupted by shutdown"); } } } @@ -260,7 +270,7 @@ public void ClearTemporaryRetryDelayBounds() /// public void Close() { - if (ReadyState != ReadyState.Raw && ReadyState != ReadyState.Shutdown) + if (ReadyState != ReadyState.Shutdown) { Close(ReadyState.Shutdown); } @@ -332,7 +342,7 @@ private async Task ConnectToEventSourceAsync(CancellationToken cancellationToken var svc = GetEventSourceService(_configuration); svc.ConnectionOpened += (o, e) => { - _lastSuccessfulConnectionTime = DateTime.Now; + _connectionTimer.Restart(); SetReadyState(ReadyState.Open, OnOpened, e.Headers); }; svc.ConnectionClosed += (o, e) => { SetReadyState(ReadyState.Closed, OnClosed); }; @@ -440,7 +450,10 @@ private void HandleParsedLine(EventParser.Result result) { if (long.TryParse(result.GetValueAsString(), out var retry)) { - _backOff.SetServerDirectedMinDelay(TimeSpan.FromMilliseconds(retry)); + // Clamp while the value is still an integer. + var millis = Math.Max(0, + Math.Min(retry, Constants.MaxServerDirectedRetryDelayMillis)); + _backOff.SetServerDirectedMinDelay(TimeSpan.FromMilliseconds(millis)); } } } diff --git a/src/LaunchDarkly.EventSource/ExponentialBackoffWithDecorrelation.cs b/src/LaunchDarkly.EventSource/ExponentialBackoffWithDecorrelation.cs index 5c8bddf..f07599d 100644 --- a/src/LaunchDarkly.EventSource/ExponentialBackoffWithDecorrelation.cs +++ b/src/LaunchDarkly.EventSource/ExponentialBackoffWithDecorrelation.cs @@ -4,8 +4,6 @@ namespace LaunchDarkly.EventSource { internal sealed class ExponentialBackoffWithDecorrelation { - internal static readonly TimeSpan MaxServerDirectedMinDelay = TimeSpan.FromHours(1); - private readonly object _lock = new object(); private readonly Random _randomSource; @@ -57,10 +55,18 @@ public TimeSpan GetNextBackOff() /// applied: minimumDelay * 2^n, limited to the maximum. /// /// + /// + /// The caller must hold the lock. This reads the bounds without taking it because it is the + /// inner core of , which already holds it; taking it here would + /// only be a re-entrant acquisition on the hot path. Anything needing a synchronized read of + /// the bounds on its own should use the accessors below, which take the lock themselves. + /// + /// /// n is compared against the ceiling rather than used to compute a value that is /// then clamped, so no intermediate ever exceeds the maximum. That matters because /// n is unbounded: a long outage keeps incrementing it, and computing /// minimumDelay * 2^n directly would overflow. + /// /// internal long GetUnjitteredMillisecondsForN(int n) { @@ -105,18 +111,14 @@ internal long GetUnjitteredMillisecondsForN(int n) /// double from wherever the progression had reached. /// /// - /// Values above are capped at it. Negative - /// values become zero. + /// Negative values become zero. No upper bound is applied here; a caller taking a value + /// from an untrusted source is responsible for capping it first. /// /// /// the server-directed reconnection time public void SetServerDirectedMinDelay(TimeSpan minDelay) { var value = NonNegative(minDelay); - if (value > MaxServerDirectedMinDelay) - { - value = MaxServerDirectedMinDelay; - } lock (_lock) { _serverDirectedMinDelay = value; diff --git a/src/LaunchDarkly.EventSource/IEventSource.cs b/src/LaunchDarkly.EventSource/IEventSource.cs index 4fd0cf3..4ec4dba 100644 --- a/src/LaunchDarkly.EventSource/IEventSource.cs +++ b/src/LaunchDarkly.EventSource/IEventSource.cs @@ -106,11 +106,19 @@ public interface IEventSource /// fail. /// /// + /// A server-directed reconnection time received via the SSE retry: field supersedes + /// , and the configured + /// along with it. Such a + /// value does not affect , so the ceiling set here continues to + /// bound reconnection. + /// + /// /// It is safe to call this from any thread, including from an /// handler. /// /// - /// the lowest delay to use while these bounds are in effect + /// the lowest delay to use while these bounds are in effect, + /// unless a server-directed retry: value supersedes it /// the highest delay to use while these bounds are in effect /// void SetTemporaryRetryDelayBounds(TimeSpan initialDelay, TimeSpan maxDelay); @@ -118,15 +126,26 @@ public interface IEventSource /// /// Discards any bounds installed by /// , restoring the configured - /// and - /// . + /// and, unless a server-directed + /// retry: value is in effect, the configured + /// . /// /// + /// /// Restoring the configured bounds resets the backoff level. If no temporary bounds /// are in effect this does nothing. The count of reconnection attempts is not reset. /// Calling this is not required as the configured bounds are also restored automatically /// after a connection stays open for /// . + /// + /// + /// A server-directed reconnection time received via the SSE retry: field is not + /// discarded and continues to supersede + /// , so the configured + /// minimum may not be the one actually in use afterwards. Only a later retry: value + /// replaces it. is unaffected + /// and is always restored. + /// /// /// void ClearTemporaryRetryDelayBounds(); diff --git a/test/LaunchDarkly.EventSource.Tests/EventSourceReconnectingTest.cs b/test/LaunchDarkly.EventSource.Tests/EventSourceReconnectingTest.cs index 996bbdd..a69b8f1 100644 --- a/test/LaunchDarkly.EventSource.Tests/EventSourceReconnectingTest.cs +++ b/test/LaunchDarkly.EventSource.Tests/EventSourceReconnectingTest.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Generic; using System.Net; +using System.Threading; using System.Threading.Tasks; using LaunchDarkly.TestHelpers.HttpTest; using Xunit; @@ -108,6 +109,70 @@ public void RetryDelayDurationsShouldIncrease() AssertBackoffsAlwaysIncrease(backoffs, nAttempts); } + [Fact] + public void CloseBeforeStartAsyncReportsShutdown() + { + // Closed fires many times over one instance -- once per connection cycle -- and + // carries the state it is reporting, so a consumer can tell a per-connection close + // from a terminal one. Closing before the stream ever started was the one path that + // shut down silently, leaving no way to observe it. + using (var es = MakeEventSource(_uri)) + { + var reported = new List(); + es.Closed += (_, e) => reported.Add(e.ReadyState); + + es.Close(); + + Assert.Equal(new[] { ReadyState.Shutdown }, reported); + Assert.Equal(ReadyState.Shutdown, es.ReadyState); + } + } + + [Fact] + public void ClosingTwiceReportsShutdownOnce() + { + using (var es = MakeEventSource(_uri)) + { + var reported = new List(); + es.Closed += (_, e) => reported.Add(e.ReadyState); + + es.Close(); + es.Close(); + + // Shutdown is terminal in SetReadyState, so the repeat is silent. + Assert.Single(reported); + } + } + + [Fact] + public async Task CloseBeforeStartAsyncDoesNotSpin() + { + // Close() from the Raw state deliberately does not transition to Shutdown, so a loop + // that tested only ReadyState would keep running -- and because Close() latches the + // shutdown token, every backoff wait would return instantly, reconnecting with no + // pacing at all. + var handler = Handlers.Status((int)HttpStatusCode.InternalServerError); + + using (var server = HttpServer.Start(handler)) + { + using (var es = MakeEventSource(server.Uri, + c => c.InitialRetryDelay(TimeSpan.FromMilliseconds(10)))) + { + var errors = 0; + es.Error += (_, __) => Interlocked.Increment(ref errors); + + es.Close(); + + var streamTask = Task.Run(es.StartAsync); + var finished = await Task.WhenAny(streamTask, + Task.Delay(TimeSpan.FromSeconds(2))); + + Assert.Same(streamTask, finished); + Assert.InRange(Volatile.Read(ref errors), 0, 1); + } + } + } + [Fact] public async Task CloseInterruptsAPendingBackoffWait() { diff --git a/test/LaunchDarkly.EventSource.Tests/ExponentialBackoffWithDecorrelationTest.cs b/test/LaunchDarkly.EventSource.Tests/ExponentialBackoffWithDecorrelationTest.cs index 90be714..b170aec 100644 --- a/test/LaunchDarkly.EventSource.Tests/ExponentialBackoffWithDecorrelationTest.cs +++ b/test/LaunchDarkly.EventSource.Tests/ExponentialBackoffWithDecorrelationTest.cs @@ -1,4 +1,6 @@ using System; +using System.Collections.Generic; +using System.Numerics; using Xunit; namespace LaunchDarkly.EventSource.Tests @@ -334,21 +336,6 @@ public void ServerDirectedMinDelayResetsN() Assert.Equal(0, expo.GetBackoffN()); } - [Fact] - public void ServerDirectedMinDelayIsClampedToOneHour() - { - // The maximum is deliberately larger than the wire cap so the cap is the bound under - // test, not the ceiling. - var expo = Seeded(Base1s, TimeSpan.FromHours(24)); - - expo.SetServerDirectedMinDelay(TimeSpan.FromHours(2)); - - Assert.Equal(ExponentialBackoffWithDecorrelation.MaxServerDirectedMinDelay, - expo.GetServerDirectedMinDelay()); - Assert.Equal((long)TimeSpan.FromHours(1).TotalMilliseconds, - expo.GetUnjitteredMillisecondsForN(0)); - } - [Fact] public void ZeroServerDirectedMinDelayProducesZeroDelay() { @@ -616,5 +603,100 @@ public void SetBoundsToNegativeIsNoOpWhenAlreadyZero() expo.SetBounds(TimeSpan.FromMilliseconds(-10), TimeSpan.FromMilliseconds(-20)); Assert.Equal(1, expo.GetBackoffN()); } + + #region Exact reference for the delay formula + + // GetUnjitteredMillisecondsForN computes min << n clamped to max, using shift tricks and an + // n >= 63 guard to avoid overflowing the 64-bit shift. These two tests check it against an + // arbitrary-precision reference that cannot overflow, so the optimisation is pinned to the + // arithmetic it is standing in for rather than to hand-picked expectations. + + private const long MaxTimeSpanMs = 922337203685477L; // TimeSpan.MaxValue in whole ms + + private static ExponentialBackoffWithDecorrelation FromMillis(long minMs, long maxMs) => + new ExponentialBackoffWithDecorrelation( + TimeSpan.FromTicks(minMs * TimeSpan.TicksPerMillisecond), + TimeSpan.FromTicks(maxMs * TimeSpan.TicksPerMillisecond)); + + private static long ExactReference(long minMs, long maxMs, int n) + { + if (minMs <= 0 || maxMs <= 0) + { + return 0; + } + var exact = new BigInteger(minMs) * BigInteger.Pow(2, n <= 0 ? 0 : n); + return exact > new BigInteger(maxMs) ? maxMs : (long)exact; + } + + [Fact] + public void DelayFormulaMatchesExactReferenceForCuratedBounds() + { + // Powers of two and their neighbours, the shipped defaults, the server-directed cap, + // and the boundaries of int and TimeSpan -- the places a shift or a cast goes wrong. + var interesting = new long[] + { + 0, 1, 2, 3, 4, 5, 7, 8, 9, 15, 16, 17, 19, 20, 31, 32, 33, 63, 64, 65, + 99, 100, 127, 128, 129, 255, 256, 999, 1000, 1023, 1024, 1025, + 29999, 30000, 30001, 65535, 65536, 3600000, 86400000, + int.MaxValue - 1L, int.MaxValue, int.MaxValue + 1L, + (1L << 31) - 1, 1L << 31, (1L << 40) + 12345, + MaxTimeSpanMs - 1, MaxTimeSpanMs + }; + + var failures = new List(); + foreach (var min in interesting) + { + foreach (var max in interesting) + { + var backoff = FromMillis(min, max); + for (var n = -3; n <= 70; n++) // negatives and n past the 63-bit guard + { + var actual = backoff.GetUnjitteredMillisecondsForN(n); + var expected = ExactReference(min, max, n); + if (actual != expected && failures.Count < 20) + { + failures.Add($"min={min} max={max} n={n} actual={actual} expected={expected}"); + } + } + } + } + Assert.True(failures.Count == 0, string.Join("; ", failures)); + } + + [Fact] + public void DelayFormulaMatchesExactReferenceForRandomizedBounds() + { + // Fixed seed, so a failure is reproducible. The count is deliberately modest: the + // reference uses BigInteger, and this runs inside the normal suite. + var random = new Random(12345); + var failures = new List(); + for (var i = 0; i < 200000; i++) + { + var min = NextInclusive(random, 0, MaxTimeSpanMs); + var max = NextInclusive(random, 0, MaxTimeSpanMs); + var backoff = FromMillis(min, max); + for (var k = 0; k < 5; k++) + { + var n = random.Next(-2, 72); + var actual = backoff.GetUnjitteredMillisecondsForN(n); + var expected = ExactReference(min, max, n); + if (actual != expected && failures.Count < 20) + { + failures.Add($"min={min} max={max} n={n} actual={actual} expected={expected}"); + } + } + } + Assert.True(failures.Count == 0, string.Join("; ", failures)); + } + + private static long NextInclusive(Random random, long minInclusive, long maxInclusive) + { + var buffer = new byte[8]; + random.NextBytes(buffer); + var value = BitConverter.ToInt64(buffer, 0) & long.MaxValue; + return minInclusive + value % (maxInclusive - minInclusive + 1); + } + + #endregion } } diff --git a/test/LaunchDarkly.EventSource.Tests/ServerDirectedRetryDelayTest.cs b/test/LaunchDarkly.EventSource.Tests/ServerDirectedRetryDelayTest.cs index 083f9b6..4c8897e 100644 --- a/test/LaunchDarkly.EventSource.Tests/ServerDirectedRetryDelayTest.cs +++ b/test/LaunchDarkly.EventSource.Tests/ServerDirectedRetryDelayTest.cs @@ -1,4 +1,5 @@ -using System; +using System; +using System.Diagnostics; using System.Threading.Tasks; using LaunchDarkly.TestHelpers.HttpTest; using Xunit; @@ -115,15 +116,97 @@ public void RetryFieldDoesNotRevertAnActiveTemporaryCeiling() }); } + // Pins the documented contract on ClearTemporaryRetryDelayBounds: it restores the + // configured ceiling but must not discard a server-directed minimum. + [Fact] + public void RetryFieldSurvivesClearTemporaryRetryDelayBounds() + { + WithRetryLine("300", es => + { + es.SetTemporaryRetryDelayBounds(TimeSpan.FromMinutes(5), TimeSpan.FromMinutes(10)); + es.ClearTemporaryRetryDelayBounds(); + + Assert.Equal(TimeSpan.FromMilliseconds(300), + es.BackOff.GetServerDirectedMinDelay()); + Assert.Equal(300L, es.BackOff.GetUnjitteredMillisecondsForN(0)); + + // The ceiling, unlike the minimum, is genuinely restored. + Assert.Equal(ConfiguredMax, es.BackOff.GetMaximumDelay()); + }); + } + [Fact] public void RetryFieldAboveOneHourIsClampedToOneHour() { // ConfiguredMax is 24 hours so the wire cap, not the ceiling, is the bound under test. WithRetryLine("7200000", es => - Assert.Equal(ExponentialBackoffWithDecorrelation.MaxServerDirectedMinDelay, + Assert.Equal(TimeSpan.FromMilliseconds(Constants.MaxServerDirectedRetryDelayMillis), es.BackOff.GetServerDirectedMinDelay())); } + // A server can send any number that parses as a long, which is a far wider range than + // TimeSpan can represent. The clamp has to happen while the value is still an integer; + // otherwise constructing the TimeSpan throws before any cap can apply, and the connection + // is torn down every time the line is read without the directive being applied at all. + [Theory] + [InlineData("1000000000000000")] // 1e15 ms, just past TimeSpan.MaxValue + [InlineData("9223372036854775807")] // long.MaxValue + public void HugeRetryFieldIsClampedRatherThanThrowing(string value) + { + WithRetryLine(value, es => + Assert.Equal(TimeSpan.FromMilliseconds(Constants.MaxServerDirectedRetryDelayMillis), + es.BackOff.GetServerDirectedMinDelay())); + } + + [Theory] + [InlineData("-1000000000000000")] + [InlineData("-9223372036854775808")] // long.MinValue + public void HugeNegativeRetryFieldIsClampedRatherThanThrowing(string value) + { + WithRetryLine(value, es => + Assert.Equal(TimeSpan.Zero, es.BackOff.GetServerDirectedMinDelay())); + } + + [Fact] + public void RetryFieldChangesTheActualWaitBeforeReconnecting() + { + var handler = Handlers.Sequential( + StartStream() + .Then(Handlers.WriteChunkString("retry: 400\n")) + .Then(WriteEvent(Marker)), // first stream then ends, forcing a reconnect + StartStream().Then(LeaveStreamOpen()) + ); + + WithServerAndEventSource(handler, + // 10ms configured initial delay: without the directed value the reconnect would + // land at roughly 5-10ms, two orders of magnitude below the assertion. + c => c.InitialRetryDelay(TimeSpan.FromMilliseconds(10)) + .MaxRetryDelay(TimeSpan.FromSeconds(30)) + .BackoffResetThreshold(TimeSpan.FromSeconds(30)), + (server, es) => + { + var sink = new EventSink(es, _testLogging); + _ = Task.Run(es.StartAsync); + + server.Recorder.RequireRequest(); + var timer = Stopwatch.StartNew(); + + // Receiving the event proves the preceding retry: line was parsed. + sink.ExpectActions( + EventSink.OpenedAction(), + EventSink.MessageReceivedAction(Marker)); + + server.Recorder.RequireRequest(); + timer.Stop(); + + // retry: 400 puts the floor at 200ms; 150ms leaves room for timer skew while + // staying far above the 10ms the configured delay would have produced. + Assert.True(timer.Elapsed >= TimeSpan.FromMilliseconds(150), + $"reconnected after {timer.ElapsedMilliseconds}ms; retry: 400 should have " + + "delayed it by at least 200ms, so the directed value did not reach the wait"); + }); + } + [Fact] public void RetryFieldOfZeroDisablesBackoff() { diff --git a/test/LaunchDarkly.EventSource.Tests/TemporaryRetryDelayBoundsTest.cs b/test/LaunchDarkly.EventSource.Tests/TemporaryRetryDelayBoundsTest.cs index ed8e161..9d06279 100644 --- a/test/LaunchDarkly.EventSource.Tests/TemporaryRetryDelayBoundsTest.cs +++ b/test/LaunchDarkly.EventSource.Tests/TemporaryRetryDelayBoundsTest.cs @@ -179,17 +179,35 @@ public void NegativeTemporaryBoundsClampToZeroWithoutThrowing() } [Fact] - public void AbsurdTemporaryBoundsCannotBreakTheReconnectLoop() + public async Task AbsurdTemporaryBoundsDoNotFaultTheReconnectLoop() { - using (var es = MakeConfigured()) - { - es.SetTemporaryRetryDelayBounds(TimeSpan.FromMilliseconds(1000), TimeSpan.MaxValue); + // Bounds this large exceed what Task.Delay accepts, and MaybeWaitWithBackOff runs + // outside the reconnect loop's exception handling -- so before the delay was clamped + // this threw ArgumentOutOfRangeException out of StartAsync and the stream was dead + // permanently. Asserting on GetNextBackOff alone cannot catch that: it is one layer + // below where the failure occurs. + var handler = Handlers.Sequential( + Handlers.Status((int)HttpStatusCode.InternalServerError), + StartStream().Then(LeaveStreamOpen()) + ); - // No caller-supplied bound may make a later delay computation throw; that would - // escape StartAsync and stop the stream for good. - for (int i = 0; i < 64; i++) + using (var server = HttpServer.Start(handler)) + { + using (var es = MakeEventSource(server.Uri, + c => c.InitialRetryDelay(TimeSpan.FromMilliseconds(10)))) { - es.BackOff.GetNextBackOff(); + es.SetTemporaryRetryDelayBounds(TimeSpan.MaxValue, TimeSpan.MaxValue); + + var streamTask = Task.Run(es.StartAsync); + server.Recorder.RequireRequest(); + + var finished = await Task.WhenAny(streamTask, + Task.Delay(TimeSpan.FromSeconds(2))); + + // Still waiting, not faulted: the clamped delay is honored as a very long + // wait rather than crashing the loop. + Assert.NotSame(streamTask, finished); + Assert.False(streamTask.IsFaulted); } } } From 3d02357469669472b2e13edd8067a04611bb4328 Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Fri, 11 Sep 2026 09:23:25 -0400 Subject: [PATCH 5/5] refactor: clamp the jitter bound with Math.Min Matches the clamping idiom already used for the server-directed retry value in EventSource, rather than expressing the same saturation as a ternary. Behaviorally identical: GetUnjitteredMillisecondsForN returns either zero or a positive value, so the result of Math.Min is always within int range and the narrowing cast cannot lose information. --- .../ExponentialBackoffWithDecorrelation.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/LaunchDarkly.EventSource/ExponentialBackoffWithDecorrelation.cs b/src/LaunchDarkly.EventSource/ExponentialBackoffWithDecorrelation.cs index f07599d..c4f4908 100644 --- a/src/LaunchDarkly.EventSource/ExponentialBackoffWithDecorrelation.cs +++ b/src/LaunchDarkly.EventSource/ExponentialBackoffWithDecorrelation.cs @@ -41,7 +41,7 @@ public TimeSpan GetNextBackOff() // Random.Next takes an int bound. 2^31 milliseconds is far longer than any // reconnect delay we would use, so saturating here cannot affect a realistic // delay. - int jitterBound = unjittered > int.MaxValue ? int.MaxValue : (int)unjittered; + int jitterBound = (int)Math.Min(unjittered, int.MaxValue); long delayMillis = unjittered / 2 + (jitterBound > 0 ? _randomSource.Next(jitterBound) / 2 : 0);