Skip to content

feat!: temporary retry-delay bounds and SSE retry: field handling #126

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tanderson-ld merged 5 commits into from
Sep 11, 2026
Merged

feat!: temporary retry-delay bounds and SSE retry: field handling #126

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
d3866c7
feat!: add temporary retry-delay bounds and apply the SSE retry: field
tanderson-ld Aug 28, 2026
b786b7d
fix: interrupt a pending backoff wait on shutdown
tanderson-ld Aug 31, 2026
6c77182
test: avoid a blocking task wait in the concurrency test
tanderson-ld Aug 31, 2026
7a39b04
fix: address first-round review findings
tanderson-ld Sep 1, 2026
3d02357
refactor: clamp the jitter bound with Math.Min
tanderson-ld Sep 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions src/LaunchDarkly.EventSource/Configuration.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,10 @@ public sealed class Configuration
/// <summary>
/// The amount of time a connection must stay open before the EventSource resets its backoff delay.
/// </summary>
/// <remarks>
/// A connection that stays open for at least this long also discards any temporary bounds set
/// by <see cref="IEventSource.SetTemporaryRetryDelayBounds(TimeSpan, TimeSpan)"/>.
/// </remarks>
/// <seealso cref="ConfigurationBuilder.BackoffResetThreshold(TimeSpan)"/>
public TimeSpan BackoffResetThreshold { get; }

Expand Down
14 changes: 13 additions & 1 deletion src/LaunchDarkly.EventSource/ConfigurationBuilder.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
using System;
using System;
using System.Collections.Generic;
using System.Net.Http;
using System.Text;
Expand Down Expand Up @@ -97,6 +97,11 @@ public ConfigurationBuilder HttpRequestModifier(Action<HttpRequestMessage> httpR
/// a backoff algorithm.
/// </para>
/// <para>
/// A server-directed reconnection time received via the SSE <c>retry:</c> field supersedes
/// this value, as do bounds installed by
/// <see cref="IEventSource.SetTemporaryRetryDelayBounds(TimeSpan, TimeSpan)"/>.
/// </para>
/// <para>
/// The default value is <see cref="Configuration.DefaultInitialRetryDelay"/>. Negative values
/// are changed to zero.
/// </para>
Expand Down Expand Up @@ -148,6 +153,13 @@ public ConfigurationBuilder MaxRetryDelay(TimeSpan maxRetryDelay)
/// value. This prevents long delays from occurring on connections that are only rarely restarted.
/// </para>
/// <para>
/// A connection that stays open for at least this long also discards any bounds set by
/// <see cref="IEventSource.SetTemporaryRetryDelayBounds(TimeSpan, TimeSpan)"/>, restoring
/// <see cref="InitialRetryDelay(TimeSpan)"/> and <see cref="MaxRetryDelay(TimeSpan)"/>. A
/// server-directed reconnection time received via the SSE <c>retry:</c> field is not
/// discarded.
/// </para>
/// <para>
/// The default value is <see cref="Configuration.DefaultBackoffResetThreshold"/>. Negative
/// values are changed to zero.
/// </para>
Expand Down
5 changes: 5 additions & 0 deletions src/LaunchDarkly.EventSource/Constants.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,11 @@ internal static class Constants
/// </summary>
internal static string RetryField = "retry";

/// <summary>
/// The largest reconnection time honored from a Server Sent Event retry field.
/// </summary>
internal const long MaxServerDirectedRetryDelayMillis = 3_600_000;

/// <summary>
/// The identifier field name in a Server Sent Event.
/// </summary>
Expand Down
78 changes: 60 additions & 18 deletions src/LaunchDarkly.EventSource/EventSource.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Net.Http;
using System.Threading;
Expand All @@ -24,10 +25,13 @@ 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 readonly CancellationTokenSource _shutdownTokenSource = new CancellationTokenSource();
private static readonly TimeSpan MaxSleepTime = TimeSpan.FromMilliseconds(int.MaxValue);
private readonly Stopwatch _connectionTimer = new Stopwatch();
private ReadyState _readyState;

#endregion
Expand Down Expand Up @@ -78,6 +82,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
Expand All @@ -94,9 +101,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();
}
Expand All @@ -121,13 +130,17 @@ 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)
{
_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;
_connectionTimer.Reset();
}
await MaybeWaitWithBackOff();
}
Expand Down Expand Up @@ -199,13 +212,26 @@ public async Task StartAsync()
}

private async Task MaybeWaitWithBackOff() {
if (_retryDelay.TotalMilliseconds > 0)
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)
{
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;
try
{
await Task.Delay(sleepTime, _shutdownTokenSource.Token);
}
catch (OperationCanceledException)
{
// Cancellation happened during the wait, likely intentional close
_logger.Debug("Backoff wait interrupted by shutdown");
}
}
}
Expand All @@ -221,22 +247,35 @@ public void Restart(bool resetBackoffDelay)
}
if (resetBackoffDelay)
{
_backOff.ResetReconnectAttemptCount();
_backOff.ResetBackoffN();
}
}
CancelCurrentRequest();
}

/// <inheritdoc/>
public void SetTemporaryRetryDelayBounds(TimeSpan initialDelay, TimeSpan maxDelay)
{
_backOff.SetBounds(initialDelay, maxDelay);
}

/// <inheritdoc/>
public void ClearTemporaryRetryDelayBounds()
{
_backOff.SetBounds(_configuredInitialRetryDelay, _configuredMaxRetryDelay);
}

/// <summary>
/// Closes the connection to the SSE server. The <c>EventSource</c> cannot be reopened after this.
/// </summary>
public void Close()
{
if (ReadyState != ReadyState.Raw && ReadyState != ReadyState.Shutdown)
if (ReadyState != ReadyState.Shutdown)
{
Close(ReadyState.Shutdown);
}
CancelCurrentRequest();
_shutdownTokenSource.Cancel();

// do not dispose httpClient if it is user provided
if (_configuration.HttpClient == null)
Expand Down Expand Up @@ -303,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); };
Expand Down Expand Up @@ -411,7 +450,10 @@ private void HandleParsedLine(EventParser.Result result)
{
if (long.TryParse(result.GetValueAsString(), out var retry))
{
_retryDelay = 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));
}
}
}
Expand Down
Loading
Loading