From 9274f7fd40d35eeb55e98c1f8b9b5ee9b7841cd4 Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Wed, 5 Aug 2026 18:11:34 -0400 Subject: [PATCH 1/3] feat: add named RetryCurve API for switching retry regimes at runtime Introduces a RetryCurve opaque handle. Callers construct curves via NewRetryCurve(options...), designate them at subscribe time via StreamOptionDefaultRetryCurve / StreamOptionRegisterRetryCurve, and switch between them at runtime via Stream.ActivateCurve. Enables SDKs to run a multi-regime retry policy (e.g., a normal regime + an extended regime for auth failures) while keeping the library's single-regime timing path intact for legacy callers. Overlay resolution walks (active-curve spec -> effective-default spec -> hard-coded fallbacks), evaluated lazily at delay-computation time. Per-curve formula counter n is retained across activations. Healthy-operation reset zeros all curves' formula counters and reverts to the effective default; it does not clear base-delay overrides (matches SSE spec's "reconnection time is set until updated"). SSE `retry:` field is honored per HTML5 semantics: the stream read loop updates every registered curve's base-delay override. Values above 1 hour are clamped per RETRY spec section 1.11.4 (new MaxServerDirectedRetryDelay constant). Clamping happens in milliseconds before the multiplication by time.Millisecond so extreme wire values cannot overflow the Duration. Internal changes: - Widened backoffStrategy.applyBackoff and jitterStrategy.applyJitter to accept per-call maxDelay / ratio so a single strategy instance can serve multiple curves. Math bodies unchanged from the pre-existing library. - Renamed internal SetBaseDelay to ApplyRetryTime; it now iterates all registered curves. Legacy stream options (StreamOptionInitialRetry / UseBackoff / UseJitter / RetryResetInterval) continue to work unchanged; when no explicit RetryCurve is provided they synthesize the effective default. Refs SDK-2788. --- contract-tests/go.sum | 4 +- retry_curve.go | 102 +++++++++ retry_curve_test.go | 495 ++++++++++++++++++++++++++++++++++++++++++ retry_delay.go | 280 ++++++++++++++++++------ retry_delay_test.go | 84 ++++++- server.go | 6 +- stream.go | 44 ++-- stream_options.go | 71 +++++- 8 files changed, 977 insertions(+), 109 deletions(-) create mode 100644 retry_curve.go create mode 100644 retry_curve_test.go diff --git a/contract-tests/go.sum b/contract-tests/go.sum index acf1872..6da908e 100644 --- a/contract-tests/go.sum +++ b/contract-tests/go.sum @@ -6,5 +6,5 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/stretchr/testify v1.6.0 h1:jlIyCplCJFULU/01vCkhKuTyc3OorI3bJFuw6obfgho= github.com/stretchr/testify v1.6.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= -gopkg.in/yaml.v3 v3.0.0 h1:hjy8E9ON/egN1tAYqKb61G10WtihqetD4sz2H+8nIeA= -gopkg.in/yaml.v3 v3.0.0/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/retry_curve.go b/retry_curve.go new file mode 100644 index 0000000..6418343 --- /dev/null +++ b/retry_curve.go @@ -0,0 +1,102 @@ +package eventsource + +import "time" + +// RetryCurve is an opaque handle to a retry curve. Construct one via NewRetryCurve, +// install it as the stream's default via StreamOptionDefaultRetryCurve, or register +// it as an additional curve via StreamOptionRegisterRetryCurve, and activate it at +// runtime via Stream.ActivateCurve. +// +// Two different NewRetryCurve calls with identical options yield two different +// curves (identity is by pointer, not by parameter equality). +// +// All three per-curve properties (base delay, max delay, jitter) may be left unset +// by omitting the corresponding RetryCurveOption. Unset properties inherit from the +// stream's effective default at delay-computation time. +// +// Reset interval (the healthy-operation threshold for returning to the effective +// default) is a stream-level concept rather than a per-curve one; configure it via +// StreamOptionRetryResetInterval. Whichever curve is currently active, the reset +// check uses the stream's single reset interval. +// +// A server-directed `retry:` hint on the SSE wire is also stream-wide per HTML5 +// semantics: it overrides every registered curve's base delay for subsequent +// attempts (clamped to MaxServerDirectedRetryDelay). The curve's declared maxDelay +// ceiling is untouched — the "always retry within maxDelay" property that motivates +// registering an extended curve is preserved even after a hint. +type RetryCurve struct { + baseDelay *time.Duration + maxDelay *time.Duration + jitter *float64 +} + +// RetryCurveOption is a common interface for configuration parameters that can be +// used when creating a RetryCurve via NewRetryCurve. This mirrors the interface-based +// StreamOption pattern used elsewhere in this package. +type RetryCurveOption interface { + apply(*RetryCurve) error +} + +type retryCurveBaseDelayOption struct{ v time.Duration } + +func (o retryCurveBaseDelayOption) apply(c *RetryCurve) error { + c.baseDelay = &o.v + return nil +} + +// RetryCurveBaseDelay returns an option that sets the base delay for a RetryCurve. +// Without this option, base delay is inherited from the stream's effective default +// at delay-computation time. +func RetryCurveBaseDelay(base time.Duration) RetryCurveOption { + return retryCurveBaseDelayOption{v: base} +} + +type retryCurveMaxDelayOption struct{ v time.Duration } + +func (o retryCurveMaxDelayOption) apply(c *RetryCurve) error { + c.maxDelay = &o.v + return nil +} + +// RetryCurveMaxDelay returns an option that sets the maximum delay (backoff ceiling) +// for a RetryCurve. A max delay of zero means "no backoff" — successive retries all +// use the base delay. Without this option, max delay is inherited from the stream's +// effective default at delay-computation time. +func RetryCurveMaxDelay(max time.Duration) RetryCurveOption { + return retryCurveMaxDelayOption{v: max} +} + +type retryCurveJitterOption struct{ v float64 } + +func (o retryCurveJitterOption) apply(c *RetryCurve) error { + c.jitter = &o.v + return nil +} + +// RetryCurveJitter returns an option that sets the jitter ratio (range 0.0 - 1.0) +// for a RetryCurve. A jitter ratio of zero means "no jitter." Without this option, +// jitter is inherited from the stream's effective default at delay-computation time. +func RetryCurveJitter(ratio float64) RetryCurveOption { + return retryCurveJitterOption{v: ratio} +} + +// NewRetryCurve constructs a RetryCurve from the provided options. Properties not +// specified via options are inherited from the stream's effective default. The library +// never mutates the fields set here; the returned pointer is a stable, opaque handle. +func NewRetryCurve(options ...RetryCurveOption) *RetryCurve { + c := &RetryCurve{} + for _, o := range options { + _ = o.apply(c) + } + return c +} + +// DefaultCurve is a package-level sentinel *RetryCurve meaning "revert to +// this stream's effective default." Pass it to Stream.ActivateCurve when you want +// to explicitly revert without holding a reference to the default handle directly +// (for example, when the default was configured via legacy stream options rather +// than via StreamOptionDefaultRetryCurve). +// +// The sentinel has no data — its purpose is pointer identity only. Do not pass it +// to StreamOption wrappers; use NewRetryCurve for that. +var DefaultCurve = &RetryCurve{} //nolint:gochecknoglobals diff --git a/retry_curve_test.go b/retry_curve_test.go new file mode 100644 index 0000000..05e0efb --- /dev/null +++ b/retry_curve_test.go @@ -0,0 +1,495 @@ +package eventsource + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +// mkRetryDelayWithCurves builds a retryDelayStrategy from streamOptions that +// include a default curve, any number of registered curves, and an optional +// stream-level reset interval. A resetInterval of 0 defers to the library default +// (DefaultRetryResetInterval). +func mkRetryDelayWithCurves( + defaultCurve *RetryCurve, + registered []*RetryCurve, + resetInterval time.Duration, + randSeed int64, +) *retryDelayStrategy { + opts := &streamOptions{ + defaultRetryCurve: defaultCurve, + registeredRetryCurves: registered, + retryResetInterval: resetInterval, + } + return newRetryDelayStrategyFromOptions(opts, randSeed) +} + +func TestActiveCurveReturnsEffectiveDefaultAtStart(t *testing.T) { + def := NewRetryCurve( + RetryCurveBaseDelay(time.Second), + RetryCurveMaxDelay(time.Second*30), + ) + r := mkRetryDelayWithCurves(def, nil, 0, 0) + assert.Same(t, def, r.activeCurve()) +} + +func TestActivateCurveSwitchesToRegisteredCurveImmediately(t *testing.T) { + def := NewRetryCurve(RetryCurveBaseDelay(time.Second)) + ext := NewRetryCurve(RetryCurveBaseDelay(time.Minute * 5)) + r := mkRetryDelayWithCurves(def, []*RetryCurve{ext}, 0, 0) + + r.activateCurve(ext) + // Activation is immediate; the observer sees the change right away. + assert.Same(t, ext, r.activeCurve()) + + d := r.NextRetryDelay(time.Now()) + assert.Same(t, ext, r.activeCurve()) + assert.Equal(t, time.Minute*5, d) +} + +func TestActivateCurveUnregisteredIsSilentNoOp(t *testing.T) { + def := NewRetryCurve(RetryCurveBaseDelay(time.Second)) + ext := NewRetryCurve(RetryCurveBaseDelay(time.Minute * 5)) + unrelated := NewRetryCurve(RetryCurveBaseDelay(time.Minute)) + r := mkRetryDelayWithCurves(def, []*RetryCurve{ext}, 0, 0) + + r.activateCurve(unrelated) + // Unrelated curve silently ignored; active still points at default. + assert.Same(t, def, r.activeCurve()) + d := r.NextRetryDelay(time.Now()) + assert.Equal(t, time.Second, d) +} + +func TestActivateCurveDefaultSentinelRevertsToEffectiveDefault(t *testing.T) { + def := NewRetryCurve(RetryCurveBaseDelay(time.Second)) + ext := NewRetryCurve(RetryCurveBaseDelay(time.Minute * 5)) + r := mkRetryDelayWithCurves(def, []*RetryCurve{ext}, 0, 0) + + // Move to extended immediately. + r.activateCurve(ext) + assert.Same(t, ext, r.activeCurve()) + + // Sentinel means "revert to effective default." + r.activateCurve(DefaultCurve) + assert.Same(t, def, r.activeCurve()) + d := r.NextRetryDelay(time.Now()) + assert.Equal(t, time.Second, d) +} + +func TestPerCurveRetryCountIsIndependent(t *testing.T) { + def := NewRetryCurve( + RetryCurveBaseDelay(time.Second), + RetryCurveMaxDelay(time.Minute), + ) + ext := NewRetryCurve( + RetryCurveBaseDelay(time.Minute*5), + RetryCurveMaxDelay(time.Hour), + ) + r := mkRetryDelayWithCurves(def, []*RetryCurve{ext}, 0, 0) + + t0 := time.Now() + // Progress default's counter to 3 attempts. + _ = r.NextRetryDelay(t0) // 1s (counter 0 → 1) + _ = r.NextRetryDelay(t0) // 2s (counter 1 → 2) + _ = r.NextRetryDelay(t0) // 4s (counter 2 → 3) + + // Switch to extended. Extended's counter is still 0; first extended delay is baseDelay. + r.activateCurve(ext) + d := r.NextRetryDelay(t0) + assert.Equal(t, time.Minute*5, d) + + // Second extended attempt: 5min * 2 = 10min. + d = r.NextRetryDelay(t0) + assert.Equal(t, time.Minute*10, d) + + // Revert to default: counter picks up where it left off (was 3, next attempt uses 3 → 8s). + r.activateCurve(DefaultCurve) + d = r.NextRetryDelay(t0) + assert.Equal(t, time.Second*8, d) +} + +// Reset trumps a same-cycle activation: a caller who activated an alternative curve +// before NextRetryDelay observes the reset condition sees the reset override their +// choice. This gives the natural semantic that a transient unexpected failure after +// a long healthy period does not push the SDK into the alternative regime. +func TestResetTrumpsActivation(t *testing.T) { + def := NewRetryCurve( + RetryCurveBaseDelay(time.Second), + RetryCurveMaxDelay(time.Minute), + ) + ext := NewRetryCurve( + RetryCurveBaseDelay(time.Minute*5), + RetryCurveMaxDelay(time.Hour), + ) + resetInterval := time.Second * 30 + r := mkRetryDelayWithCurves(def, []*RetryCurve{ext}, resetInterval, 0) + + t0 := time.Now() + // Establish a healthy period longer than resetInterval. + r.SetGoodSince(t0) + + // Caller activates ext (immediate). + r.activateCurve(ext) + assert.Same(t, ext, r.activeCurve()) + + // NextRetryDelay fires with a currentTime past the reset threshold. Reset trumps + // activation: active reverts to default, counters zeroed, delay uses default.baseDelay. + d := r.NextRetryDelay(t0.Add(resetInterval)) + assert.Same(t, def, r.activeCurve()) + assert.Equal(t, time.Second, d) +} + +// Persistent failures — after reset trumps the first activation, subsequent activations +// (with no intervening healthy period long enough to reset) succeed and drive extended +// regime progression. +func TestPersistentFailuresRampIntoExtended(t *testing.T) { + def := NewRetryCurve( + RetryCurveBaseDelay(time.Second), + RetryCurveMaxDelay(time.Minute), + ) + ext := NewRetryCurve( + RetryCurveBaseDelay(time.Minute*5), + RetryCurveMaxDelay(time.Hour), + ) + resetInterval := time.Second * 30 + r := mkRetryDelayWithCurves(def, []*RetryCurve{ext}, resetInterval, 0) + + t0 := time.Now() + r.SetGoodSince(t0) + + // Failure #1: after healthy period. Reset trumps activation. + r.activateCurve(ext) + d := r.NextRetryDelay(t0.Add(resetInterval)) + assert.Equal(t, time.Second, d) // reset fired, default active + + // Failure #2 shortly after: no healthy period, no reset. + r.activateCurve(ext) + d = r.NextRetryDelay(t0.Add(resetInterval + time.Millisecond)) + assert.Same(t, ext, r.activeCurve()) + assert.Equal(t, time.Minute*5, d) + + // Failure #3: continues extended progression. + r.activateCurve(ext) // no-op, ext already active + d = r.NextRetryDelay(t0.Add(resetInterval + time.Millisecond*2)) + assert.Equal(t, time.Minute*10, d) +} + +func TestResetZerosAllCurvesCounters(t *testing.T) { + def := NewRetryCurve( + RetryCurveBaseDelay(time.Second), + RetryCurveMaxDelay(time.Minute), + ) + ext := NewRetryCurve( + RetryCurveBaseDelay(time.Minute*5), + RetryCurveMaxDelay(time.Hour), + ) + resetInterval := time.Second * 30 + r := mkRetryDelayWithCurves(def, []*RetryCurve{ext}, resetInterval, 0) + + t0 := time.Now() + + // Progress default. + _ = r.NextRetryDelay(t0) + _ = r.NextRetryDelay(t0) + + // Progress extended. + r.activateCurve(ext) + _ = r.NextRetryDelay(t0) + _ = r.NextRetryDelay(t0) + + // Return to default, healthy period elapses, then failure triggers reset. + r.activateCurve(DefaultCurve) + r.SetGoodSince(t0) + d := r.NextRetryDelay(t0.Add(resetInterval)) + assert.Equal(t, time.Second, d) // default's counter zeroed + + // Confirm extended's counter was also zeroed. + r.activateCurve(ext) + d = r.NextRetryDelay(t0.Add(resetInterval + time.Millisecond)) + assert.Equal(t, time.Minute*5, d) +} + +// ApplyRetryTime is stream-level per the SSE spec's "reconnection time" semantics. +// It updates every registered curve's baseDelay uniformly and resets each +// curve's formula counter, so subsequent attempts in any regime start from the +// hinted base. Each regime still enforces its own maxDelay ceiling. +func TestApplyRetryTimeUpdatesAllCurves(t *testing.T) { + def := NewRetryCurve( + RetryCurveBaseDelay(time.Second), + RetryCurveMaxDelay(time.Second*30), + ) + ext := NewRetryCurve( + RetryCurveBaseDelay(time.Minute*5), + RetryCurveMaxDelay(time.Hour), + ) + r := mkRetryDelayWithCurves(def, []*RetryCurve{ext}, 0, 0) + + t0 := time.Now() + + // Progress default's counter so we can verify the reset that comes with the + // hint. + _ = r.NextRetryDelay(t0) + _ = r.NextRetryDelay(t0) + + // Server hint: 500ms. All curves' baseDelays should become 500ms and their + // formula counters should be zeroed. + r.ApplyRetryTime(time.Millisecond * 500) + + // Default: first attempt after hint uses the hinted value literally. + d := r.NextRetryDelay(t0) + assert.Equal(t, time.Millisecond*500, d) + + // Switch to extended. Extended's baseDelay is now also 500ms (per stream-level + // mutation) but its maxDelay ceiling of 1hr still applies. + r.activateCurve(ext) + d = r.NextRetryDelay(t0) + assert.Equal(t, time.Millisecond*500, d) +} + +// Base-delay mutations by ApplyRetryTime persist across healthy-operation reset — +// matching the SSE spec's "reconnection time is set until updated" semantic. +// Reset zeros counters and reverts the active pointer, but does not touch the +// mutated base delays. +func TestBaseDelayMutationPersistsAcrossReset(t *testing.T) { + def := NewRetryCurve(RetryCurveBaseDelay(time.Second)) + resetInterval := time.Second * 30 + r := mkRetryDelayWithCurves(def, nil, resetInterval, 0) + + t0 := time.Now() + + // Server hints 500ms. + r.ApplyRetryTime(time.Millisecond * 500) + d := r.NextRetryDelay(t0) + assert.Equal(t, time.Millisecond*500, d) + + // Enter healthy state, then trigger reset. + r.SetGoodSince(t0.Add(time.Second)) + d = r.NextRetryDelay(t0.Add(time.Second + resetInterval)) + + // Reset zeroed the counter, but the 500ms baseDelay persists. + assert.Equal(t, time.Millisecond*500, d) +} + +func TestOverlayInheritanceFillsUnsetPropertiesFromDefault(t *testing.T) { + def := NewRetryCurve( + RetryCurveBaseDelay(time.Second), + RetryCurveMaxDelay(time.Minute), + RetryCurveJitter(0), + ) + // ext specifies only baseDelay + maxDelay; jitter should inherit from default (0). + ext := NewRetryCurve( + RetryCurveBaseDelay(time.Minute*5), + RetryCurveMaxDelay(time.Hour), + ) + r := mkRetryDelayWithCurves(def, []*RetryCurve{ext}, 0, 0) + + // Verify extended's resolved values by observing behavior. First extended attempt + // yields 5min (baseDelay) with no jitter (inherited 0). + r.activateCurve(ext) + d := r.NextRetryDelay(time.Now()) + assert.Equal(t, time.Minute*5, d) +} + +func TestNilRegisteredCurvesAreIgnored(t *testing.T) { + def := NewRetryCurve(RetryCurveBaseDelay(time.Second)) + r := mkRetryDelayWithCurves(def, []*RetryCurve{nil, nil}, 0, 0) + // Should not panic and should still work as a single-curve stream. + d := r.NextRetryDelay(time.Now()) + assert.Equal(t, time.Second, d) +} + +func TestRegisteredCurveEqualToDefaultIsDeduped(t *testing.T) { + def := NewRetryCurve(RetryCurveBaseDelay(time.Second)) + // Registering the default as an additional curve is a no-op. + r := mkRetryDelayWithCurves(def, []*RetryCurve{def}, 0, 0) + assert.Len(t, r.curves, 1) +} + +func TestLegacyOptionsSynthesizeEffectiveDefault(t *testing.T) { + // No explicit default curve provided; legacy stream options should populate + // the effective default. + opts := &streamOptions{ + initialRetry: time.Millisecond * 500, + backoffMaxDelay: time.Second * 10, + jitterRatio: 0, + retryResetInterval: time.Second * 30, + } + r := newRetryDelayStrategyFromOptions(opts, 0) + + // First delay uses baseDelay from legacy options. + d := r.NextRetryDelay(time.Now()) + assert.Equal(t, time.Millisecond*500, d) + // Second delay applies backoff: 500ms * 2 = 1s. + d = r.NextRetryDelay(time.Now()) + assert.Equal(t, time.Second, d) +} + +func TestNoConfigurationYieldsHardCodedFallbackBehavior(t *testing.T) { + // No default curve, no legacy option overrides. The library synthesizes an + // empty *RetryCurve as the effective default; overlay resolution falls through + // to hard-coded fallbacks (DefaultInitialRetry for baseDelay; no backoff; no + // jitter). Verify by observing behavior: first delay is DefaultInitialRetry. + opts := &streamOptions{} + r := newRetryDelayStrategyFromOptions(opts, 0) + d := r.NextRetryDelay(time.Now()) + assert.Equal(t, DefaultInitialRetry, d) +} + +// The extended regime targeted by the RETRY spec starts at 5 minutes and doubles +// until it clamps to a 1-hour ceiling: 5m, 10m, 20m, 40m, 1hr, 1hr, ... This test +// pins the exact sequence so the epic-stated behavior is guarded against +// accidental math regressions. +func TestExtendedCurveProgressionMatchesRetrySpec(t *testing.T) { + ext := NewRetryCurve( + RetryCurveBaseDelay(time.Minute*5), + RetryCurveMaxDelay(time.Hour), + ) + r := mkRetryDelayWithCurves(ext, nil, 0, 0) + + expected := []time.Duration{ + time.Minute * 5, + time.Minute * 10, + time.Minute * 20, + time.Minute * 40, + time.Hour, + time.Hour, + time.Hour, + } + t0 := time.Now() + for i, want := range expected { + got := r.NextRetryDelay(t0) + assert.Equal(t, want, got, "attempt %d", i) + } +} + +// The public StreamOption wrappers must build the same strategy shape as +// constructing the streamOptions struct directly. This is the only test that +// exercises defaultRetryCurveOption.apply and registerRetryCurveOption.apply +// end-to-end; every other curve-related test bypasses the options by calling +// mkRetryDelayWithCurves. +func TestStreamOptionCurveWrappersWireStrategy(t *testing.T) { + def := NewRetryCurve(RetryCurveBaseDelay(time.Second)) + ext := NewRetryCurve( + RetryCurveBaseDelay(time.Minute*5), + RetryCurveMaxDelay(time.Hour), + ) + + opts := &streamOptions{} + assert.NoError(t, StreamOptionDefaultRetryCurve(def).apply(opts)) + assert.NoError(t, StreamOptionRegisterRetryCurve(ext).apply(opts)) + r := newRetryDelayStrategyFromOptions(opts, 0) + + // Effective default is the curve installed via the option. + assert.Same(t, def, r.effectiveDefault) + // First delay uses def's baseDelay. + assert.Equal(t, time.Second, r.NextRetryDelay(time.Now())) + // ext is reachable via activation. + r.activateCurve(ext) + assert.Equal(t, time.Minute*5, r.NextRetryDelay(time.Now())) +} + +// When both a legacy StreamOptionInitialRetry and a new StreamOptionDefaultRetryCurve +// are provided, the explicit curve wins and the legacy value is silently ignored. +// This pins the currently-undocumented precedence so a future change to it is a +// deliberate act. +func TestExplicitDefaultCurveOverridesLegacyInitialRetry(t *testing.T) { + explicit := NewRetryCurve(RetryCurveBaseDelay(time.Millisecond * 250)) + + opts := &streamOptions{} + assert.NoError(t, StreamOptionInitialRetry(time.Second*7).apply(opts)) + assert.NoError(t, StreamOptionDefaultRetryCurve(explicit).apply(opts)) + r := newRetryDelayStrategyFromOptions(opts, 0) + + // The explicit curve's baseDelay wins; the legacy 7s value is ignored. + assert.Equal(t, time.Millisecond*250, r.NextRetryDelay(time.Now())) +} + +func TestActivateCurveNilIsSilentNoOp(t *testing.T) { + def := NewRetryCurve(RetryCurveBaseDelay(time.Second)) + ext := NewRetryCurve(RetryCurveBaseDelay(time.Minute * 5)) + r := mkRetryDelayWithCurves(def, []*RetryCurve{ext}, 0, 0) + + r.activateCurve(ext) + assert.Same(t, ext, r.activeCurve()) + + // activateCurve(nil) must not panic and must not change the active pointer. + r.activateCurve(nil) + assert.Same(t, ext, r.activeCurve()) +} + +// With more than one registered curve, healthy-op reset must zero every curve's +// retryCount (not just the active one or the default), and ApplyRetryTime must +// set every curve's baseDelayOverride. Existing tests only exercise a single +// registered curve; this one guards the loop against silently mis-iterating. +func TestMultipleRegisteredCurvesAllTrackedByResetAndApplyRetryTime(t *testing.T) { + def := NewRetryCurve(RetryCurveBaseDelay(time.Second), RetryCurveMaxDelay(time.Minute)) + extA := NewRetryCurve(RetryCurveBaseDelay(time.Minute*5), RetryCurveMaxDelay(time.Hour)) + extB := NewRetryCurve(RetryCurveBaseDelay(time.Minute*15), RetryCurveMaxDelay(time.Hour)) + resetInterval := time.Second * 30 + r := mkRetryDelayWithCurves(def, []*RetryCurve{extA, extB}, resetInterval, 0) + + t0 := time.Now() + + // Progress each curve's counter. + _ = r.NextRetryDelay(t0) // def: 1s (n 0→1) + r.activateCurve(extA) + _ = r.NextRetryDelay(t0) // extA: 5m (n 0→1) + r.activateCurve(extB) + _ = r.NextRetryDelay(t0) // extB: 15m (n 0→1) + + // Reset. All three counters must be zero. + r.activateCurve(DefaultCurve) + r.SetGoodSince(t0) + _ = r.NextRetryDelay(t0.Add(resetInterval)) + + // Confirm every curve's counter was zeroed by activating each and observing + // the first delay equals its declared baseDelay (n=0 branch). + r.activateCurve(extA) + assert.Equal(t, time.Minute*5, r.NextRetryDelay(t0.Add(resetInterval))) + r.activateCurve(extB) + assert.Equal(t, time.Minute*15, r.NextRetryDelay(t0.Add(resetInterval))) + + // ApplyRetryTime must hit every curve's baseDelayOverride. + r.ApplyRetryTime(time.Millisecond * 750) + r.activateCurve(DefaultCurve) + assert.Equal(t, time.Millisecond*750, r.NextRetryDelay(t0.Add(resetInterval))) + r.activateCurve(extA) + assert.Equal(t, time.Millisecond*750, r.NextRetryDelay(t0.Add(resetInterval))) + r.activateCurve(extB) + assert.Equal(t, time.Millisecond*750, r.NextRetryDelay(t0.Add(resetInterval))) +} + +// Registering the same curve pointer multiple times is deduped down to a single +// entry in the curves map. Complements TestRegisteredCurveEqualToDefaultIsDeduped, +// which covers the (default, registered) collision; this covers the +// (registered, registered) collision. +func TestDuplicateRegisteredCurveIsDeduped(t *testing.T) { + def := NewRetryCurve(RetryCurveBaseDelay(time.Second)) + ext := NewRetryCurve(RetryCurveBaseDelay(time.Minute * 5)) + + r := mkRetryDelayWithCurves(def, []*RetryCurve{ext, ext, ext}, 0, 0) + assert.Len(t, r.curves, 2) // def + ext, not def + ext + ext + ext +} + +// RetryCurveBaseDelay(0) is a legitimate way to say "zero base delay" and is +// semantically distinct from "unset" — an unset baseDelay falls through overlay +// resolution to the effective default (or the hard-coded fallback). Explicit +// zero, in contrast, produces a 0 duration for the first attempt. This test +// pins the distinction. +func TestRetryCurveBaseDelayZeroIsExplicitNotUnset(t *testing.T) { + // Effective default has a nonzero base, so if `explicit-zero` were treated + // as `unset`, the first delay would be 1s (from the default) rather than 0. + def := NewRetryCurve(RetryCurveBaseDelay(time.Second)) + explicitZero := NewRetryCurve(RetryCurveBaseDelay(0)) + r := mkRetryDelayWithCurves(def, []*RetryCurve{explicitZero}, 0, 0) + + r.activateCurve(explicitZero) + assert.Equal(t, time.Duration(0), r.NextRetryDelay(time.Now())) + + // Sanity: an unset baseDelay on a different curve DOES fall through to def. + unset := NewRetryCurve(RetryCurveMaxDelay(time.Second * 10)) + r2 := mkRetryDelayWithCurves(def, []*RetryCurve{unset}, 0, 0) + r2.activateCurve(unset) + assert.Equal(t, time.Second, r2.NextRetryDelay(time.Now())) +} diff --git a/retry_delay.go b/retry_delay.go index 305dc1f..57f6dab 100644 --- a/retry_delay.go +++ b/retry_delay.go @@ -7,42 +7,57 @@ import ( "time" ) -// Encapsulation of configurable backoff/jitter behavior. +// Encapsulation of the streaming retry-timing behavior. Supports one or more +// RetryCurves registered on a stream at subscribe time, with a single +// currently-active curve driving delay computation. See retry_curve.go for the +// user-facing type; this file holds the internal machinery. // -// - The system can either be in a "good" state or a "bad" state. The initial state is "bad"; the -// caller is responsible for indicating when it transitions to "good". When we ask for a new retry -// delay, that implies the state is now transitioning to "bad". +// The library uses lazy resolution at delay-computation time, using the active curve's spec, +// falling through to the effective default's spec and then to hard-coded +// fallbacks. // -// - There is a configurable base delay, which can be changed at any time (if the SSE server sends -// us a "retry:" directive). +// Per-curve runtime state carries only two things: `retryCount` (the backoff +// formula counter n, per RETRY spec) and a nullable `baseDelayOverride` that +// captures server-directed `retry:` hints. // -// - There are optional strategies for applying backoff and jitter to the delay. -// -// This object is meant to be used from a single goroutine once it's been created; its methods are -// not safe for concurrent use. +// `baseDelayOverride` is NOT cleared on healthy-op reset — it persists until the +// server sends another `retry:` hint or the Stream is closed, matching the HTML5 +// SSE spec's "reconnection time is set until updated" semantic. type retryDelayStrategy struct { - baseDelay time.Duration - backoff backoffStrategy - jitter jitterStrategy - resetInterval time.Duration - retryCount int - goodSince time.Time // nonzero only if the state is currently "good" - lock sync.Mutex + curves map[*RetryCurve]*perCurveState + effectiveDefault *RetryCurve + active *RetryCurve + resetInterval time.Duration + goodSince time.Time // nonzero only if the state is currently "good" + backoff backoffStrategy + jitter jitterStrategy + lock sync.Mutex +} + +// perCurveState carries the per-stream mutable runtime state for one registered +// curve: its backoff-formula counter and any server-directed base-delay override. +// The curve's own spec fields (baseDelay/maxDelay/jitter) live on the *RetryCurve +// key itself and are read only. +type perCurveState struct { + retryCount int + baseDelayOverride *time.Duration } -// Abstraction for backoff delay behavior. +// Abstraction for backoff delay behavior. The per-attempt effective maxDelay is +// passed in per call so a single strategy instance can serve multiple retry curves +// with different ceilings. type backoffStrategy interface { - applyBackoff(baseDelay time.Duration, retryCount int) time.Duration + applyBackoff(baseDelay time.Duration, retryCount int, maxDelay time.Duration) time.Duration } -// Abstraction for delay jitter behavior. +// Abstraction for delay jitter behavior. The per-attempt effective ratio is passed +// in per call so a single strategy instance can serve multiple retry curves with +// different jitter ratios. type jitterStrategy interface { - applyJitter(computedDelay time.Duration) time.Duration + applyJitter(computedDelay time.Duration, ratio float64) time.Duration } -type defaultBackoffStrategy struct { - maxDelay time.Duration -} +type defaultBackoffStrategy struct{} // Creates the default implementation of exponential backoff, which doubles the delay each time up to // the specified maximum. @@ -50,74 +65,163 @@ type defaultBackoffStrategy struct { // If a resetInterval was specified for the retryDelayStrategy, and the system has been in a "good" // state for at least that long, the delay is reset back to the base. This avoids perpetually increasing // delays in a situation where failures are rare). -func newDefaultBackoff(maxDelay time.Duration) backoffStrategy { - return defaultBackoffStrategy{maxDelay} +func newDefaultBackoff() backoffStrategy { + return defaultBackoffStrategy{} } -func (s defaultBackoffStrategy) applyBackoff(baseDelay time.Duration, retryCount int) time.Duration { - d := math.Min(float64(baseDelay)*math.Pow(2, float64(retryCount)), float64(s.maxDelay)) +func (s defaultBackoffStrategy) applyBackoff(baseDelay time.Duration, retryCount int, maxDelay time.Duration) time.Duration { + d := math.Min(float64(baseDelay)*math.Pow(2, float64(retryCount)), float64(maxDelay)) return time.Duration(d) } type defaultJitterStrategy struct { - ratio float64 random *rand.Rand } // Creates the default implementation of jitter, which subtracts a pseudo-random amount from each delay. -// The ratio parameter should be greater than 0 and less than or equal to 1.0. -func newDefaultJitter(ratio float64, randSeed int64) jitterStrategy { +func newDefaultJitter(randSeed int64) jitterStrategy { if randSeed <= 0 { randSeed = time.Now().UnixNano() } - if ratio > 1.0 { - ratio = 1.0 - } //nolint:gosec // This isn't a cryptographic use-case, weak RNG is acceptable - return &defaultJitterStrategy{ratio, rand.New(rand.NewSource(randSeed))} + return &defaultJitterStrategy{random: rand.New(rand.NewSource(randSeed))} } -func (s *defaultJitterStrategy) applyJitter(computedDelay time.Duration) time.Duration { - // retryCount doesn't matter here - it's included in the int - jitter := time.Duration(s.random.Int63n(int64(float64(computedDelay) * s.ratio))) +func (s *defaultJitterStrategy) applyJitter(computedDelay time.Duration, ratio float64) time.Duration { + if ratio > 1.0 { + ratio = 1.0 + } + jitter := time.Duration(s.random.Int63n(int64(float64(computedDelay) * ratio))) return computedDelay - jitter } -// Creates a retryDelayStrategy. -func newRetryDelayStrategy( - baseDelay time.Duration, - resetInterval time.Duration, - backoff backoffStrategy, - jitter jitterStrategy, -) *retryDelayStrategy { +// newRetryDelayStrategyFromOptions constructs a retryDelayStrategy from resolved +// streamOptions. +func newRetryDelayStrategyFromOptions(opts *streamOptions, randSeed int64) *retryDelayStrategy { + // Resolve the effective default curve. + effectiveDefault := opts.defaultRetryCurve + if effectiveDefault == nil { + // Synthesize from legacy stream options. + synth := &RetryCurve{} + if opts.initialRetry > 0 { + v := opts.initialRetry + synth.baseDelay = &v + } + if opts.backoffMaxDelay > 0 { + v := opts.backoffMaxDelay + synth.maxDelay = &v + } + if opts.jitterRatio > 0 { + v := opts.jitterRatio + synth.jitter = &v + } + effectiveDefault = synth + } + + // Build the per-curve runtime state map: effective default + any additional + // registered curves. + curves := map[*RetryCurve]*perCurveState{ + effectiveDefault: {}, + } + for _, c := range opts.registeredRetryCurves { + if c == nil || c == effectiveDefault { + continue + } + if _, dup := curves[c]; dup { + continue + } + curves[c] = &perCurveState{} + } + + // Stream-level reset interval falls back to the library default. + resetInterval := opts.retryResetInterval + if resetInterval <= 0 { + resetInterval = DefaultRetryResetInterval + } + return &retryDelayStrategy{ - baseDelay: baseDelay, - resetInterval: resetInterval, - backoff: backoff, - jitter: jitter, + curves: curves, + effectiveDefault: effectiveDefault, + active: effectiveDefault, + resetInterval: resetInterval, + backoff: newDefaultBackoff(), + jitter: newDefaultJitter(randSeed), + } +} + +// firstNonNil walks the two curve layers (primary then secondary) and returns the +// value of the first whose selected field is non-nil. Falls back to `fallback` if +// neither has the field set (or is itself nil). +func firstNonNil[T any](selector func(*RetryCurve) *T, primary, secondary *RetryCurve, fallback T) T { + if primary != nil { + if v := selector(primary); v != nil { + return *v + } + } + if secondary != nil { + if v := selector(secondary); v != nil { + return *v + } } + return fallback +} + +// resolveCurveProperties computes the effective (baseDelay, maxDelay, jitter) for the given +// curve handle by walking the overlay stack: curve.spec → effectiveDefault.spec → +// hard-coded fallbacks. +// +// Caller must hold r.lock. +func (r *retryDelayStrategy) resolveCurveProperties(c *RetryCurve) (baseDelay, maxDelay time.Duration, jitter float64) { + baseDelay = firstNonNil(func(c *RetryCurve) *time.Duration { return c.baseDelay }, c, r.effectiveDefault, DefaultInitialRetry) + maxDelay = firstNonNil(func(c *RetryCurve) *time.Duration { return c.maxDelay }, c, r.effectiveDefault, time.Duration(0)) + jitter = firstNonNil(func(c *RetryCurve) *float64 { return c.jitter }, c, r.effectiveDefault, float64(0)) + return } -// NextRetryDelay computes the next retry interval. This also sets the current state to "bad". +// NextRetryDelay computes the next retry interval and marks the current state as "bad". // -// Note that currentTime is passed as a parameter instead of computed by this function to guarantee predictable -// behavior in tests. +// Order of operations: +// 1. Check the healthy-operation reset condition (goodSince non-zero AND elapsed >= +// resetInterval). If satisfied, apply reset: for each curve set retryCount = 0 +// (do NOT touch baseDelayOverride — persistent per SSE spec); +// active = effective default. +// 2. Clear goodSince +// 3. Resolve the active curve's effective properties; baseDelayOverride, if set, wins. +// 4. Compute delay = min(effectiveBase * 2^retryCount, effectiveMax) with jitter. +// 5. Increment active.retryCount. +// +// currentTime is passed as a parameter rather than computed internally to keep tests +// deterministic. func (r *retryDelayStrategy) NextRetryDelay(currentTime time.Time) time.Duration { r.lock.Lock() defer r.lock.Unlock() if !r.goodSince.IsZero() && r.resetInterval > 0 && (currentTime.Sub(r.goodSince) >= r.resetInterval) { - r.retryCount = 0 + for _, c := range r.curves { + c.retryCount = 0 + // baseDelayOverride is NOT cleared intentionally + } + r.active = r.effectiveDefault } r.goodSince = time.Time{} - delay := r.baseDelay - if r.backoff != nil { - delay = r.backoff.applyBackoff(delay, r.retryCount) + + activeState := r.curves[r.active] + + effectiveBase, effectiveMax, effectiveJitter := r.resolveCurveProperties(r.active) + if activeState.baseDelayOverride != nil { + effectiveBase = *activeState.baseDelayOverride } - r.retryCount++ - if r.jitter != nil { - delay = r.jitter.applyJitter(delay) + + delay := effectiveBase + if effectiveMax > 0 { + delay = r.backoff.applyBackoff(effectiveBase, activeState.retryCount, effectiveMax) + } + if effectiveJitter > 0 { + delay = r.jitter.applyJitter(delay, effectiveJitter) } + + activeState.retryCount++ + return delay } @@ -128,19 +232,57 @@ func (r *retryDelayStrategy) SetGoodSince(goodSince time.Time) { r.lock.Unlock() } -// SetBaseDelay changes the initial retry delay and resets the backoff (if any) so the next retry will use -// that value. +// ApplyRetryTime records a server-directed reconnection-time hint received via the +// SSE `retry:` field. It sets every registered curve's baseDelayOverride to the +// given duration and zeroes every curve's retryCount (the backoff-formula counter), +// so the immediate next attempt in whatever regime is active uses the hinted value. +// The active curve is NOT changed. +func (r *retryDelayStrategy) ApplyRetryTime(hint time.Duration) { + r.lock.Lock() + defer r.lock.Unlock() + for _, c := range r.curves { + v := hint + c.baseDelayOverride = &v + c.retryCount = 0 + } +} + +// activateCurve switches the currently-active curve immediately. Silent no-op if +// the curve is nil, unregistered, or already active. // -// This is used to implement the optional SSE behavior where the server sends a "retry:" command to -// set the base retry to a specific value. Note that we will still apply a jitter, if jitter is enabled, -// and subsequent retries will still increase exponentially. -func (r *retryDelayStrategy) SetBaseDelay(baseDelay time.Duration) { +// Does NOT reset the newly-activated curve's retryCount — each curve's counter +// retains its progression across activations. Does NOT touch any curve's +// baseDelayOverride. +// +// If a healthy-operation reset fires on the next NextRetryDelay call, that reset +// trumps this activation: active will be reverted to the effective default. +func (r *retryDelayStrategy) activateCurve(c *RetryCurve) { r.lock.Lock() - r.baseDelay = baseDelay - r.retryCount = 0 - r.lock.Unlock() + defer r.lock.Unlock() + if c == nil { + return + } + if c == DefaultCurve { + r.active = r.effectiveDefault + return + } + if _, ok := r.curves[c]; !ok { + return + } + r.active = c +} + +// activeCurve returns the currently-active *RetryCurve — a real curve pointer +// registered on this stream, never the DefaultCurve sentinel. +func (r *retryDelayStrategy) activeCurve() *RetryCurve { + r.lock.Lock() + defer r.lock.Unlock() + return r.active } func (r *retryDelayStrategy) hasJitter() bool { //nolint:unused // used only in tests - return r.jitter != nil + r.lock.Lock() + defer r.lock.Unlock() + _, _, j := r.resolveCurveProperties(r.active) + return j > 0 } diff --git a/retry_delay_test.go b/retry_delay_test.go index 62056ef..138c8af 100644 --- a/retry_delay_test.go +++ b/retry_delay_test.go @@ -7,9 +7,30 @@ import ( "github.com/stretchr/testify/assert" ) +// mkRetryDelay constructs a retryDelayStrategy for tests that only exercise the +// (single default curve) legacy shape. Uses legacy stream-option fields, so the +// effective default is synthesized from those values, with any remaining unset +// properties falling through to the library's hard-coded fallbacks during lazy +// overlay resolution. +func mkRetryDelay( + baseDelay time.Duration, + resetInterval time.Duration, + backoffMaxDelay time.Duration, + jitterRatio float64, + randSeed int64, +) *retryDelayStrategy { + opts := &streamOptions{ + initialRetry: baseDelay, + backoffMaxDelay: backoffMaxDelay, + jitterRatio: jitterRatio, + retryResetInterval: resetInterval, + } + return newRetryDelayStrategyFromOptions(opts, randSeed) +} + func TestFixedRetryDelay(t *testing.T) { d0 := time.Second * 10 - r := newRetryDelayStrategy(d0, 0, nil, nil) + r := mkRetryDelay(d0, 0, 0, 0, 0) t0 := time.Now().Add(-time.Minute) d1 := r.NextRetryDelay(t0) d2 := r.NextRetryDelay(t0.Add(time.Second)) @@ -22,7 +43,7 @@ func TestFixedRetryDelay(t *testing.T) { func TestBackoffWithoutJitter(t *testing.T) { d0 := time.Second * 10 max := time.Minute - r := newRetryDelayStrategy(d0, 0, newDefaultBackoff(max), nil) + r := mkRetryDelay(d0, 0, max, 0, 0) t0 := time.Now().Add(-time.Minute) d1 := r.NextRetryDelay(t0) d2 := r.NextRetryDelay(t0.Add(time.Second)) @@ -37,7 +58,7 @@ func TestBackoffWithoutJitter(t *testing.T) { func TestJitterWithoutBackoff(t *testing.T) { d0 := time.Second seed := int64(1000) - r := newRetryDelayStrategy(d0, 0, nil, newDefaultJitter(0.5, seed)) + r := mkRetryDelay(d0, 0, 0, 0.5, seed) t0 := time.Now().Add(-time.Minute) d1 := r.NextRetryDelay(t0) d2 := r.NextRetryDelay(t0.Add(time.Second)) @@ -51,7 +72,7 @@ func TestJitterWithBackoff(t *testing.T) { d0 := time.Second max := time.Minute seed := int64(1000) - r := newRetryDelayStrategy(d0, 0, newDefaultBackoff(max), newDefaultJitter(0.5, seed)) + r := mkRetryDelay(d0, 0, max, 0.5, seed) t0 := time.Now().Add(-time.Minute) d1 := r.NextRetryDelay(t0) d2 := r.NextRetryDelay(t0.Add(time.Second)) @@ -65,7 +86,7 @@ func TestBackoffResetInterval(t *testing.T) { d0 := time.Second * 10 max := time.Minute resetInterval := time.Second * 45 - r := newRetryDelayStrategy(d0, resetInterval, newDefaultBackoff(max), nil) + r := mkRetryDelay(d0, resetInterval, max, 0, 0) t0 := time.Now().Add(-time.Minute) r.SetGoodSince(t0) @@ -97,10 +118,55 @@ func TestBackoffAndJitterWorkWithHighRetryCount(t *testing.T) { max := 365 * 200 * 24 * time.Hour // 200 years retryCount := 35 // 2^35 seconds exceeds a 63-bit count of nanoseconds - backoff := newDefaultBackoff(max) - jitter := newDefaultJitter(0.5, 1) + backoff := newDefaultBackoff() + jitter := newDefaultJitter(1) - d1 := backoff.applyBackoff(d0, retryCount) - _ = jitter.applyJitter(d1) + d1 := backoff.applyBackoff(d0, retryCount, max) + _ = jitter.applyJitter(d1, 0.5) // No assertion - the test just needs to not panic. } + +// RETRY spec §1.11.4: a server-directed wait duration MUST NOT exceed 1 hour; +// values above 1 hour are treated as 1 hour. The clamp is applied by +// clampServerDirectedRetry at the SSE wire boundary, before ApplyRetryTime is +// called. This test pins the clamp behavior, including that clamping happens +// before the multiplication by time.Millisecond so extreme int64 wire values +// cannot overflow the Duration. +func TestClampServerDirectedRetry(t *testing.T) { + // Below the ceiling: pass through unchanged. + assert.Equal(t, time.Millisecond*500, clampServerDirectedRetry(500)) + assert.Equal(t, time.Second*30, clampServerDirectedRetry(30_000)) + + // At the ceiling: passes through unchanged. + assert.Equal(t, MaxServerDirectedRetryDelay, + clampServerDirectedRetry(int64(MaxServerDirectedRetryDelay/time.Millisecond))) + + // Above the ceiling: clamped. + assert.Equal(t, MaxServerDirectedRetryDelay, + clampServerDirectedRetry(int64(MaxServerDirectedRetryDelay/time.Millisecond)+1)) + assert.Equal(t, MaxServerDirectedRetryDelay, clampServerDirectedRetry(9_223_372_036_854_775)) +} + +// ApplyRetryTime implements the HTML5 SSE spec's `retry:` directive: it sets the +// stream-level reconnection time. The library maps that to (a) setting each +// registered curve's baseDelayOverride uniformly and (b) resetting each curve's +// backoff-formula counter so the immediate next attempt uses the hinted value +// literally. +func TestApplyRetryTimeUpdatesBaseAndResetsFormulaCounter(t *testing.T) { + d0 := time.Second + max := time.Minute + r := mkRetryDelay(d0, 0, max, 0, 0) + + t0 := time.Now() + // Progress the counter through a few backoff steps. + _ = r.NextRetryDelay(t0) // retryCount 0 → 1 + _ = r.NextRetryDelay(t0) // retryCount 1 → 2 + _ = r.NextRetryDelay(t0) // retryCount 2 → 3 + + // Server hint: new base delay. Counter must reset so the next attempt uses the + // hint value literally (matching browser EventSource semantics). + r.ApplyRetryTime(time.Second * 2) + + d := r.NextRetryDelay(t0) + assert.Equal(t, time.Second*2, d) +} diff --git a/server.go b/server.go index 56a25e5..bff1335 100644 --- a/server.go +++ b/server.go @@ -223,12 +223,12 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { // discarding intermediate values is a safe operation. var delayedEvent eventOrComment - jitterStrategy := newDefaultJitter(0.5, 0) + jitterStrategy := newDefaultJitter(0) usingJitter := srv.jitter > 0 var jitterTimer timer if usingJitter { - jitterTimer = &goTimer{timer: time.NewTimer(jitterStrategy.applyJitter(srv.jitter))} + jitterTimer = &goTimer{timer: time.NewTimer(jitterStrategy.applyJitter(srv.jitter, 0.5))} jitterTimer.Stop() } else { jitterTimer = &noopTimer{C: make(<-chan time.Time)} @@ -298,7 +298,7 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { // Figure out the jitter and start the timer. Once this trigger, we // will write the event and clear the way for a new event to come in. - delay := jitterStrategy.applyJitter(srv.jitter) + delay := jitterStrategy.applyJitter(srv.jitter, 0.5) jitterTimer.Reset(delay) case ev, ok := <-readBatchCh: diff --git a/stream.go b/stream.go index 416e113..3c970b4 100644 --- a/stream.go +++ b/stream.go @@ -161,20 +161,7 @@ func SubscribeWithRequestAndOptions(request *http.Request, options ...StreamOpti } func newStream(request *http.Request, configuredOptions streamOptions) *Stream { - var backoff backoffStrategy - var jitter jitterStrategy - if configuredOptions.backoffMaxDelay > 0 { - backoff = newDefaultBackoff(configuredOptions.backoffMaxDelay) - } - if configuredOptions.jitterRatio > 0 { - jitter = newDefaultJitter(configuredOptions.jitterRatio, 0) - } - retryDelay := newRetryDelayStrategy( - configuredOptions.initialRetry, - configuredOptions.retryResetInterval, - backoff, - jitter, - ) + retryDelay := newRetryDelayStrategyFromOptions(&configuredOptions, 0) stream := &Stream{ c: configuredOptions.httpClient, @@ -360,7 +347,7 @@ NewStream: case ev := <-events: pub := ev.(*publication) if pub.Retry() > 0 { - stream.retryDelay.SetBaseDelay(time.Duration(pub.Retry()) * time.Millisecond) + stream.retryDelay.ApplyRetryTime(clampServerDirectedRetry(pub.Retry())) } stream.lastEventID = pub.lastEventID stream.retryDelay.SetGoodSince(time.Now()) @@ -394,6 +381,25 @@ func (stream *Stream) getRetryDelayStrategy() *retryDelayStrategy { //nolint:unu return stream.retryDelay } +// ActivateCurve switches the currently-active retry curve on this stream. The +// change takes effect immediately. +// +// The curve argument must be one of: (a) a curve registered on this stream via +// StreamOptionRegisterRetryCurve, (b) the stream's effective default curve +// (installed via StreamOptionDefaultRetryCurve, or otherwise synthesized), or +// (c) the package-level DefaultCurve sentinel — treated as a symbolic marker +// meaning "revert to the effective default." Any other *RetryCurve is a silent no-op. +// +// If a healthy-operation reset fires on the next NextRetryDelay call, that reset +// trumps this activation: the active curve will be reverted to the effective +// default. This preserves the intuitive property that a single failure after a long +// healthy period does not push the stream into the newly-activated regime. +// +// Safe to call from any goroutine, including the stream's error handler. +func (stream *Stream) ActivateCurve(curve *RetryCurve) { + stream.retryDelay.activateCurve(curve) +} + // SetLogger sets the Logger field in a thread-safe manner. func (stream *Stream) SetLogger(logger Logger) { stream.mu.Lock() @@ -406,3 +412,11 @@ func (stream *Stream) getLogger() Logger { defer stream.mu.RUnlock() return stream.Logger } + +func clampServerDirectedRetry(hintMs int64) time.Duration { + maxMs := int64(MaxServerDirectedRetryDelay / time.Millisecond) + if hintMs > maxMs { + hintMs = maxMs + } + return time.Duration(hintMs) * time.Millisecond +} diff --git a/stream_options.go b/stream_options.go index f81935a..adc9a02 100644 --- a/stream_options.go +++ b/stream_options.go @@ -7,17 +7,19 @@ import ( ) type streamOptions struct { - initialRetry time.Duration - httpClient *http.Client - lastEventID string - logger Logger - backoffMaxDelay time.Duration - jitterRatio float64 - readTimeout time.Duration - retryResetInterval time.Duration - initialRetryTimeout time.Duration - errorHandler StreamErrorHandler - queryParamsFunc *func(existing url.Values) url.Values + initialRetry time.Duration + httpClient *http.Client + lastEventID string + logger Logger + backoffMaxDelay time.Duration + jitterRatio float64 + readTimeout time.Duration + retryResetInterval time.Duration + initialRetryTimeout time.Duration + errorHandler StreamErrorHandler + queryParamsFunc *func(existing url.Values) url.Values + defaultRetryCurve *RetryCurve + registeredRetryCurves []*RetryCurve } // StreamOption is a common interface for optional configuration parameters that can be @@ -227,6 +229,49 @@ func StreamOptionLogger(logger Logger) StreamOption { return loggerOption{logger: logger} } +type defaultRetryCurveOption struct { + curve *RetryCurve +} + +func (o defaultRetryCurveOption) apply(s *streamOptions) error { + s.defaultRetryCurve = o.curve + return nil +} + +// StreamOptionDefaultRetryCurve returns an option that installs the effective default +// retry curve for the stream — the curve that is active at stream start and the +// curve the stream reverts to after a healthy-operation reset. +// +// Every stream has an effective default at all times, so reset always has a valid +// curve to revert to. If this option is not provided, the effective default is +// synthesized from the legacy stream options (StreamOptionInitialRetry, +// StreamOptionUseBackoff, StreamOptionUseJitter, StreamOptionRetryResetInterval) — +// any properties that remain unset fall through to the library's hard-coded +// fallbacks during delay-computation time. +func StreamOptionDefaultRetryCurve(curve *RetryCurve) StreamOption { + return defaultRetryCurveOption{curve: curve} +} + +type registerRetryCurveOption struct { + curve *RetryCurve +} + +func (o registerRetryCurveOption) apply(s *streamOptions) error { + if o.curve != nil { + s.registeredRetryCurves = append(s.registeredRetryCurves, o.curve) + } + return nil +} + +// StreamOptionRegisterRetryCurve returns an option that registers a curve on the +// stream, making it eligible for runtime activation via Stream.ActivateCurve. +// Unspecified properties on the curve inherit from the effective default curve. +// +// May be called multiple times to register more than one additional curve. +func StreamOptionRegisterRetryCurve(curve *RetryCurve) StreamOption { + return registerRetryCurveOption{curve: curve} +} + type streamErrorHandlerOption struct { handler StreamErrorHandler } @@ -258,4 +303,8 @@ const ( DefaultInitialRetry = time.Second * 3 // DefaultRetryResetInterval is the default value for StreamOptionRetryResetInterval. DefaultRetryResetInterval = time.Second * 60 + // MaxServerDirectedRetryDelay is the upper bound applied to server-directed + // reconnection times received via the SSE `retry:` field, per RETRY spec + // Requirement 1.11.4. Values above this ceiling are treated as this ceiling. + MaxServerDirectedRetryDelay = time.Hour * 1 ) From 206f0383b67aa028a00152a587a3de0e6aa5ee32 Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Thu, 6 Aug 2026 09:55:15 -0400 Subject: [PATCH 2/3] chore: satisfy golangci-lint on retry-curve additions - Rename `max` parameter on RetryCurveMaxDelay to `maxDelay` (revive redefines-builtin-id: `max` shadows the Go 1.21 built-in). - Add `//nolint:unused // used only in tests` to activeCurve, matching the existing convention on hasJitter. - Wrap the applyBackoff signature and the three firstNonNil calls in resolveCurveProperties across multiple lines (lll: 120-char limit). No logic changes. `make lint` and `go test ./...` both clean locally. --- retry_curve.go | 4 ++-- retry_delay.go | 25 +++++++++++++++++++------ 2 files changed, 21 insertions(+), 8 deletions(-) diff --git a/retry_curve.go b/retry_curve.go index 6418343..3de5ef2 100644 --- a/retry_curve.go +++ b/retry_curve.go @@ -62,8 +62,8 @@ func (o retryCurveMaxDelayOption) apply(c *RetryCurve) error { // for a RetryCurve. A max delay of zero means "no backoff" — successive retries all // use the base delay. Without this option, max delay is inherited from the stream's // effective default at delay-computation time. -func RetryCurveMaxDelay(max time.Duration) RetryCurveOption { - return retryCurveMaxDelayOption{v: max} +func RetryCurveMaxDelay(maxDelay time.Duration) RetryCurveOption { + return retryCurveMaxDelayOption{v: maxDelay} } type retryCurveJitterOption struct{ v float64 } diff --git a/retry_delay.go b/retry_delay.go index 57f6dab..1cc0c58 100644 --- a/retry_delay.go +++ b/retry_delay.go @@ -69,7 +69,9 @@ func newDefaultBackoff() backoffStrategy { return defaultBackoffStrategy{} } -func (s defaultBackoffStrategy) applyBackoff(baseDelay time.Duration, retryCount int, maxDelay time.Duration) time.Duration { +func (s defaultBackoffStrategy) applyBackoff( + baseDelay time.Duration, retryCount int, maxDelay time.Duration, +) time.Duration { d := math.Min(float64(baseDelay)*math.Pow(2, float64(retryCount)), float64(maxDelay)) return time.Duration(d) } @@ -171,10 +173,21 @@ func firstNonNil[T any](selector func(*RetryCurve) *T, primary, secondary *Retry // hard-coded fallbacks. // // Caller must hold r.lock. -func (r *retryDelayStrategy) resolveCurveProperties(c *RetryCurve) (baseDelay, maxDelay time.Duration, jitter float64) { - baseDelay = firstNonNil(func(c *RetryCurve) *time.Duration { return c.baseDelay }, c, r.effectiveDefault, DefaultInitialRetry) - maxDelay = firstNonNil(func(c *RetryCurve) *time.Duration { return c.maxDelay }, c, r.effectiveDefault, time.Duration(0)) - jitter = firstNonNil(func(c *RetryCurve) *float64 { return c.jitter }, c, r.effectiveDefault, float64(0)) +func (r *retryDelayStrategy) resolveCurveProperties( + c *RetryCurve, +) (baseDelay, maxDelay time.Duration, jitter float64) { + baseDelay = firstNonNil( + func(c *RetryCurve) *time.Duration { return c.baseDelay }, + c, r.effectiveDefault, DefaultInitialRetry, + ) + maxDelay = firstNonNil( + func(c *RetryCurve) *time.Duration { return c.maxDelay }, + c, r.effectiveDefault, time.Duration(0), + ) + jitter = firstNonNil( + func(c *RetryCurve) *float64 { return c.jitter }, + c, r.effectiveDefault, float64(0), + ) return } @@ -274,7 +287,7 @@ func (r *retryDelayStrategy) activateCurve(c *RetryCurve) { // activeCurve returns the currently-active *RetryCurve — a real curve pointer // registered on this stream, never the DefaultCurve sentinel. -func (r *retryDelayStrategy) activeCurve() *RetryCurve { +func (r *retryDelayStrategy) activeCurve() *RetryCurve { //nolint:unused // used only in tests r.lock.Lock() defer r.lock.Unlock() return r.active From ac1294cb1290f4e7424177605c2acd2443a1ed42 Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Thu, 6 Aug 2026 10:39:57 -0400 Subject: [PATCH 3/3] chore: revert unrelated contract-tests/go.sum bump --- contract-tests/go.sum | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/contract-tests/go.sum b/contract-tests/go.sum index 6da908e..acf1872 100644 --- a/contract-tests/go.sum +++ b/contract-tests/go.sum @@ -6,5 +6,5 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/stretchr/testify v1.6.0 h1:jlIyCplCJFULU/01vCkhKuTyc3OorI3bJFuw6obfgho= github.com/stretchr/testify v1.6.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= -gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= -gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.0 h1:hjy8E9ON/egN1tAYqKb61G10WtihqetD4sz2H+8nIeA= +gopkg.in/yaml.v3 v3.0.0/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=