From bbbb6a4b0354c9dfac798ca636e5f601f8b56d01 Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Tue, 8 Sep 2026 16:07:21 -0400 Subject: [PATCH 1/4] feat: Keep retrying a rejected auto-configuration key A 401 or 403 on the auto-configuration stream stopped the stream and exited the process. Every Relay Proxy instance in a fleet did this at once, and recovery needed an operator to restart each one, even when a persistent cache held a configuration those instances could have served from. The stream now classifies each failure and keeps retrying. A rejected key moves it to delays that grow from five minutes to one hour, so it recovers on its own once the key becomes valid again without adding load to a service that is rejecting every request. Ordinary failures keep the delays they already had, so an outage behaves as it did before. What happens to SDK requests depends on whether Relay has a configuration. With cached configuration it serves those environments and keeps trying the key indefinitely. With no configuration it can serve nothing, so it reports the failure and lets the process exit, which surfaces a bad key to whatever supervises Relay. Relay decides that as soon as it knows both facts: the key was rejected, and the cache read finished without data. The cache goroutine closes its channel when it completes, so no waiting is needed to learn there is nothing cached. Neither cache store sets its own deadline, so initTimeout bounds the wait in case the read does not complete. Setting ignoreConnectionErrors keeps Relay running and retrying instead of reporting the failure, which is what that option already promises. Move the stream off the single-regime eventsource backoff options onto retry profiles, which is what allows the second set of delays. The normal profile keeps the previous values, which already matched the retry specification. TestNoReconnectAfterUnrecoverableHTTPError described the old behavior and is replaced by four tests: the stream recovers once the key works, Relay gives up when nothing is cached, Relay stays up when the cache supplies a configuration, and ignoreConnectionErrors keeps it running with neither. --- docs/proxy-mode.md | 14 + internal/autoconfig/errors_and_messages.go | 14 +- internal/autoconfig/stream_manager.go | 241 ++++++++++++++---- .../autoconfig/stream_manager_errors_test.go | 114 ++++++++- .../stream_manager_test_base_test.go | 2 + .../reload_restart_redis_test.go | 5 +- relay/relay.go | 18 +- 7 files changed, 331 insertions(+), 77 deletions(-) diff --git a/docs/proxy-mode.md b/docs/proxy-mode.md index d3f3316e4..af77226d5 100644 --- a/docs/proxy-mode.md +++ b/docs/proxy-mode.md @@ -52,6 +52,20 @@ Versions of the Relay Proxy before 8.22.0 treated a rejected key as permanent. T If you're an Enterprise customer using [automatic configuration](https://docs.launchdarkly.com/home/advanced/relay-proxy-enterprise/automatic-configuration), the first thing the Relay Proxy does on startup is request the configuration data from LaunchDarkly. During this time, the Relay Proxy does not yet know what the configured environments are, so it has no way to know if an SDK key or other credential in a request is valid. Therefore it returns a `503` error for all requests, indicating that it isn't ready yet. In this case, all LaunchDarkly SDKs will retry after a backoff delay. +### Relay Proxy receives a request when LaunchDarkly has rejected its auto-configuration key + +If LaunchDarkly rejects the auto-configuration stream with a `401` or `403` error, the auto-configuration key is not valid. The Relay Proxy does not give up. It keeps retrying on a slower schedule, which backs off to as long as one hour between attempts, so the connection recovers on its own if the key becomes valid again. + +What happens to SDK requests in the meantime depends on whether the Relay Proxy has a configuration: + +* If you configure a [persistent store](./persistent-storage.md) and it holds configuration data from a previous run, the Relay Proxy loads that data and serves those environments while it keeps retrying. It logs `AutoConfig loaded from persistent cache`. + +* If the Relay Proxy has no configuration from any source, it cannot serve any request, because it does not know what its environments are. It shuts down, so that whatever supervises the process reports the problem. This happens as soon as the Relay Proxy has both been rejected and finished reading the persistent store, which is usually immediate. If the store does not answer, the Relay Proxy waits no longer than [`initTimeout`](./configuration.md#file-section-main) before shutting down. + + To keep the Relay Proxy running and retrying in this case instead, set [`ignoreConnectionErrors`](./configuration.md#file-section-main) to `true`. Note that until the key becomes valid, the Relay Proxy still answers every request with a `503` error, because it has no environments to serve. + +Earlier versions of the Relay Proxy shut down immediately when LaunchDarkly rejected the auto-configuration key, even when a persistent store held usable configuration data. + ### Relay Proxy receives a request with invalid credentials If a server-side or mobile SDK connects to the Relay Proxy with an invalid SDK key or mobile key, the response is a `401` error. diff --git a/internal/autoconfig/errors_and_messages.go b/internal/autoconfig/errors_and_messages.go index 97101b060..58e1a11a7 100644 --- a/internal/autoconfig/errors_and_messages.go +++ b/internal/autoconfig/errors_and_messages.go @@ -1,11 +1,15 @@ package autoconfig const ( - logMsgStreamConnecting = "Connecting to auto-configuration stream (%s)" - logMsgStreamHTTPError = "HTTP error %d on auto-configuration stream" - logMsgBadURL = "Couldn't construct auto-configuration URL: %v" - logMsgStreamOtherError = "Unexpected error on auto-configuration stream: %s" - logMsgBadKey = "Invalid auto-configuration key; cannot get environments" + logMsgStreamConnecting = "Connecting to auto-configuration stream (%s)" + logMsgStreamHTTPError = "HTTP error %d on auto-configuration stream" + logMsgBadURL = "Couldn't construct auto-configuration URL: %v" + logMsgStreamOtherError = "Unexpected error on auto-configuration stream: %s" + logMsgBadKey = "Invalid auto-configuration key; cannot get environments" + logMsgBadKeyWillRetry = "Invalid auto-configuration key; will keep retrying in case it becomes valid" + logMsgExtendedBackoff = "Classified failure as UNEXPECTED; engaging extended backoff." + logMsgNoConfigGaveUp = "Cannot get environments and no cached configuration is available; " + + "Relay cannot serve requests. Set ignoreConnectionErrors to keep Relay running and retrying" logMsgDeliberateReconnect = "Will restart auto-configuration stream to get new data due to a policy change" logMsgPutEvent = "Received configuration for %d environment(s)" logMsgAddEnv = "Added %s" diff --git a/internal/autoconfig/stream_manager.go b/internal/autoconfig/stream_manager.go index b92ca99e2..70ec193a8 100644 --- a/internal/autoconfig/stream_manager.go +++ b/internal/autoconfig/stream_manager.go @@ -19,6 +19,7 @@ import ( "github.com/launchdarkly/ld-relay/v8/config" "github.com/launchdarkly/ld-relay/v8/internal/envfactory" "github.com/launchdarkly/ld-relay/v8/internal/httpconfig" + "github.com/launchdarkly/ld-relay/v8/internal/retry" ) const ( @@ -29,6 +30,14 @@ const ( streamRetryResetInterval = 60 * time.Second streamJitterRatio = 0.5 defaultStreamRetryDelay = 1 * time.Second + + // Delays for a failure that is unlikely to correct itself soon, such as a rejected + // auto-configuration key. The stream keeps retrying on these instead of giving up, + // because an operator can make the key valid again without Relay knowing. The ceiling + // bounds how much load a fleet of Relay Proxy instances puts on a service that is + // rejecting every request. + streamExtendedRetryDelay = 5 * time.Minute + streamExtendedMaxRetryDelay = 1 * time.Hour ) var ( @@ -78,10 +87,18 @@ type StreamManager struct { lastKnownEnvs map[config.EnvironmentID]envfactory.EnvironmentRep httpConfig httpconfig.HTTPConfig initialRetryDelay time.Duration - loggers ldlog.Loggers - halt chan struct{} - done chan struct{} // closed when the subscribe goroutine exits - closeOnce sync.Once + // extendedRetryDelay is the base delay used once the service has rejected the key. + extendedRetryDelay time.Duration + // initTimeout bounds how long Relay waits for a configuration once the service has + // rejected the key. It is the initTimeout configuration option. + initTimeout time.Duration + // ignoreConnectionErrors keeps Relay running with no configuration rather than reporting + // a failure. It is the ignoreConnectionErrors configuration option. + ignoreConnectionErrors bool + loggers ldlog.Loggers + halt chan struct{} + done chan struct{} // closed when the subscribe goroutine exits + closeOnce sync.Once // cacheCh receives the result of the async cache read started by Start(). // It is consumed by consumeStream and nilled out after use. @@ -111,6 +128,8 @@ func NewStreamManager( protocolVersion int, loggers ldlog.Loggers, cache Cache, + initTimeout time.Duration, + ignoreConnectionErrors bool, ) *StreamManager { loggers.SetPrefix("AutoConfiguration") if protocolVersion > 1 { @@ -126,8 +145,13 @@ func NewStreamManager( lastKnownEnvs: make(map[config.EnvironmentID]envfactory.EnvironmentRep), httpConfig: httpConfig, initialRetryDelay: initialRetryDelay, - loggers: loggers, - halt: make(chan struct{}), + // The extended delay has no configuration key. Its value bounds the load a fleet of + // Relay Proxy instances puts on a service that is rejecting its key. + extendedRetryDelay: streamExtendedRetryDelay, + initTimeout: initTimeout, + ignoreConnectionErrors: ignoreConnectionErrors, + loggers: loggers, + halt: make(chan struct{}), status: StreamStatus{ State: interfaces.DataSourceStateInitializing, StateSince: time.Now(), @@ -218,44 +242,26 @@ func (s *StreamManager) subscribe(readyCh chan<- error) { var readyOnce sync.Once signalReady := func(err error) { readyOnce.Do(func() { readyCh <- err }) } - errorHandler := func(err error) es.StreamErrorHandlerResult { - // If Close() has been called, stop retrying so the SSE goroutine can exit. - select { - case <-s.halt: - return es.StreamErrorHandlerResult{CloseNow: true} - default: - } - - if se, ok := err.(es.SubscriptionError); ok { - errorInfo := interfaces.DataSourceErrorInfo{ - Kind: interfaces.DataSourceErrorKindErrorResponse, - StatusCode: se.Code, - Time: time.Now(), - } - if se.Code == 401 || se.Code == 403 { - s.loggers.Error(logMsgBadKey) - s.updateStatus(interfaces.DataSourceStateOff, errorInfo) - signalReady(errors.New("invalid auto-configuration key")) - return es.StreamErrorHandlerResult{CloseNow: true} - } - s.loggers.Warnf(logMsgStreamHTTPError, se.Code) - s.updateStatus(interfaces.DataSourceStateInterrupted, errorInfo) - return es.StreamErrorHandlerResult{CloseNow: false} - } - - s.loggers.Warnf(logMsgStreamOtherError, err) - s.updateStatus(interfaces.DataSourceStateInterrupted, interfaces.DataSourceErrorInfo{ - Kind: interfaces.DataSourceErrorKindNetworkError, - Message: err.Error(), - Time: time.Now(), - }) - return es.StreamErrorHandlerResult{CloseNow: false} + retryDelay := s.initialRetryDelay + if retryDelay <= 0 { + retryDelay = defaultStreamRetryDelay // COVERAGE: never happens in unit tests } - retry := s.initialRetryDelay - if retry <= 0 { - retry = defaultStreamRetryDelay // COVERAGE: never happens in unit tests - } + normalProfile := es.NewRetryProfile( + es.RetryProfileBaseDelay(retryDelay), + es.RetryProfileMaxDelay(streamMaxRetryDelay), + es.RetryProfileJitter(streamJitterRatio), + ) + extendedProfile := es.NewRetryProfile( + es.RetryProfileBaseDelay(s.extendedRetryDelay), + es.RetryProfileMaxDelay(streamExtendedMaxRetryDelay), + es.RetryProfileJitter(streamJitterRatio), + ) + + // authFailureCh tells the wait loop below that the service rejected the key. The send + // never blocks, and one notification is enough. + authFailureCh := make(chan struct{}, 1) + errorHandler := s.newStreamErrorHandler(authFailureCh, extendedProfile) rpacEndpoint, err := url.JoinPath(s.uri.String(), autoConfigStreamPath) if err != nil { @@ -279,9 +285,8 @@ func (s *StreamManager) subscribe(readyCh chan<- error) { stream, err := es.SubscribeWithRequestAndOptions(req, es.StreamOptionHTTPClient(client), es.StreamOptionReadTimeout(streamReadTimeout), - es.StreamOptionInitialRetry(retry), - es.StreamOptionUseBackoff(streamMaxRetryDelay), - es.StreamOptionUseJitter(streamJitterRatio), + es.StreamOptionDefaultRetryProfile(normalProfile), + es.StreamOptionRegisterRetryProfile(extendedProfile), es.StreamOptionRetryResetInterval(streamRetryResetInterval), es.StreamOptionErrorHandler(errorHandler), es.StreamOptionCanRetryFirstConnection(-1), @@ -293,18 +298,75 @@ func (s *StreamManager) subscribe(readyCh chan<- error) { // Race the cache read against the stream connection. If the cache returns before // the stream's first PUT, its data is applied so Relay can serve immediately. // The cache is only cancelled when a PUT arrives with authoritative data. + // Relay cannot serve a request until it knows its environments. Only the cache can supply + // them before the stream connects, so these three facts decide whether waiting is still + // worthwhile. + haveConfiguration := false // the cache supplied a configuration + cacheReported := false // the cache read finished, with or without data + authFailed := false // the service rejected the key + + // shouldGiveUp reports whether Relay has established that it cannot serve. That needs a + // rejected key and a finished cache read that produced nothing. Any other failure leaves + // Relay waiting and retrying, which is what it did before this change. + shouldGiveUp := func() bool { + return authFailed && cacheReported && !haveConfiguration && !s.ignoreConnectionErrors + } + + giveUp := func() { + s.loggers.Error(logMsgNoConfigGaveUp) + signalReady(errors.New("invalid auto-configuration key")) + s.abandonStreamGoroutine(streamCh) + } + + // initTimeoutCh stays nil, and so blocks forever in the select, until the service rejects + // the key. It bounds the wait for the cache read, which has no deadline of its own: + // neither cache store sets one, and both rely on their driver's network timeouts. + var initTimer *time.Timer + var initTimeoutCh <-chan time.Time + defer func() { + if initTimer != nil { + initTimer.Stop() + } + }() + var stream *es.Stream for stream == nil { select { case content, ok := <-s.cacheCh: + cacheReported = true if ok && content != nil { s.applyCachedContent(content) + haveConfiguration = true } if s.cacheCancel != nil { s.cacheCancel() s.cacheCancel = nil } s.cacheCh = nil + if shouldGiveUp() { + giveUp() + return + } + + case <-authFailureCh: + authFailed = true + if shouldGiveUp() { + giveUp() + return + } + if initTimer == nil && !s.ignoreConnectionErrors { + initTimer = time.NewTimer(s.initTimeout) + initTimeoutCh = initTimer.C + } + + case <-initTimeoutCh: + initTimeoutCh = nil + if !haveConfiguration { + // The cache never reported. Waiting longer cannot help, because the key is + // rejected and there is nothing to serve. + giveUp() + return + } case result := <-streamCh: if result.err != nil { @@ -320,14 +382,7 @@ func (s *StreamManager) subscribe(readyCh chan<- error) { s.cacheCancel = nil s.cacheCh = nil } - // The SSE goroutine may still be running. Drain its result in the - // background: if it produced a stream, close it so nothing leaks. - go func() { - result := <-streamCh - if result.stream != nil { - result.stream.Close() - } - }() + s.abandonStreamGoroutine(streamCh) return } } @@ -549,6 +604,86 @@ func (s *StreamManager) dispatchFilterAction(id config.FilterID, rep envfactory. } } +// newStreamErrorHandler builds the SSE error handler for one subscribe cycle. +// +// No response and no transport failure stops the stream. A rejected key can become valid +// again without Relay knowing, so a failure that is unlikely to correct itself soon moves the +// stream to the longer delays and it keeps trying. The handler reports such a failure on +// authFailureCh, because only the caller can decide whether Relay is able to serve meanwhile. +func (s *StreamManager) newStreamErrorHandler( + authFailureCh chan<- struct{}, + extendedProfile *es.RetryProfile, +) func(error) es.StreamErrorHandlerResult { + // loggedExtended keeps the notice about the longer delays to once per subscribe cycle. + // The library returns to the normal delays itself once the connection has been healthy + // for streamRetryResetInterval, and does not report that, so re-logging would mislead. + loggedExtended := false + + return func(err error) es.StreamErrorHandlerResult { + // If Close() has been called, stop retrying so the SSE goroutine can exit. + select { + case <-s.halt: + return es.StreamErrorHandlerResult{CloseNow: true} + default: + } + + class := s.classifyAndLogStreamError(err) + + result := es.StreamErrorHandlerResult{CloseNow: false} + if class == retry.Unexpected { + if !loggedExtended { + s.loggers.Info(logMsgExtendedBackoff) + loggedExtended = true + } + result.ActivateProfile = extendedProfile + select { + case authFailureCh <- struct{}{}: + default: + } + } + return result + } +} + +// classifyAndLogStreamError sorts a stream failure into a retry class and logs it. A failure +// that is unlikely to correct itself soon is worth an error, because it nearly always means a +// real configuration problem, even though the stream recovers on its own once it is fixed. +func (s *StreamManager) classifyAndLogStreamError(err error) retry.FailureClass { + var se es.SubscriptionError + if !errors.As(err, &se) { + class := retry.ClassifyTransportError(err) + if class == retry.Unexpected { + s.loggers.Errorf(logMsgStreamOtherError, err) + } else { + s.loggers.Warnf(logMsgStreamOtherError, err) + } + return class + } + + class := retry.ClassifyHTTPStatus(se.Code) + switch { + case se.Code == 401 || se.Code == 403: + s.loggers.Error(logMsgBadKeyWillRetry) + case class == retry.Unexpected: + s.loggers.Errorf(logMsgStreamHTTPError, se.Code) + default: + s.loggers.Warnf(logMsgStreamHTTPError, se.Code) + } + return class +} + +// abandonStreamGoroutine leaves the SSE connection attempt behind. That goroutine may still +// be retrying, so its result is drained in the background and any stream it produced is +// closed, which keeps the connection and its goroutines from leaking. +func (s *StreamManager) abandonStreamGoroutine(streamCh <-chan streamResult) { + go func() { + result := <-streamCh + if result.stream != nil { + result.stream.Close() + } + }() +} + func (s *StreamManager) applyCachedContent(content *PutContent) { s.handlePut(PutContent{ Environments: content.Environments, diff --git a/internal/autoconfig/stream_manager_errors_test.go b/internal/autoconfig/stream_manager_errors_test.go index 19fdcd1d2..8c15a70bb 100644 --- a/internal/autoconfig/stream_manager_errors_test.go +++ b/internal/autoconfig/stream_manager_errors_test.go @@ -1,6 +1,7 @@ package autoconfig import ( + "context" "fmt" "net/http" "testing" @@ -336,29 +337,116 @@ func TestReconnectAfterNetworkError(t *testing.T) { errorShouldCauseReconnect(t, httphelpers.BrokenConnectionHandler(), "Unexpected error") } -func TestNoReconnectAfterUnrecoverableHTTPError(t *testing.T) { +func TestRecoversAfterUnrecoverableHTTPError(t *testing.T) { + // A rejected key no longer stops the stream. An operator can make the key valid again + // without Relay knowing, so the stream keeps trying and recovers on its own, which + // previously took a process restart. for _, status := range []int{401, 403} { t.Run(fmt.Sprintf("status %d", status), func(t *testing.T) { initialEvent := makeEnvPutEvent(testEnv1) streamHandler, stream := httphelpers.SSEHandler(&initialEvent) defer stream.Close() - errorProducingHandler := httphelpers.HandlerWithStatus(status) handler := httphelpers.SequentialHandler( - errorProducingHandler, // first request will get this - streamHandler, // request after reconnect will get this + httphelpers.HandlerWithStatus(status), // first request is rejected + streamHandler, // the retry succeeds ) streamManagerTestWithStreamHandler(t, handler, stream, noopTestCache{}, func(p streamManagerTestParams) { + // Shorten the extended delay so the retry happens within the test. + p.streamManager.extendedRetryDelay = time.Millisecond + // Recovery is only observable while Relay is still running. With no cached + // configuration Relay would otherwise report failure as soon as it learned the + // cache was empty, since it has nothing to serve; the cached case is covered by + // TestKeepsRunningWhenUnauthorizedButCacheHasConfiguration. + p.streamManager.ignoreConnectionErrors = true + p.startStream() - <-p.requestsCh // first request - select { - case <-p.requestsCh: // got expected stream restart - require.Fail(t, "got unexpected stream restart") - case <-p.messageHandler.received: - require.Fail(t, "got unexpected event") - case <-time.After(time.Millisecond * 200): - p.mockLog.AssertMessageMatch(t, true, ldlog.Error, "Invalid auto-configuration key") - } + + <-p.requestsCh // the rejected request + <-p.requestsCh // the retry + + p.requireMessage() // the environment from the recovered stream + p.requireReceivedAllMessage() + + p.mockLog.AssertMessageMatch(t, true, ldlog.Error, "will keep retrying") + p.mockLog.AssertMessageMatch(t, true, ldlog.Info, "engaging extended backoff") }) }) } } + +func TestGivesUpWhenUnauthorizedAndNothingIsCached(t *testing.T) { + // With no configuration from any source, Relay can serve nothing, so it reports the + // failure after a grace period and lets the process exit. That surfaces a bad key to + // whatever supervises Relay rather than leaving a process that answers every request + // with an error. + handler := httphelpers.HandlerWithStatus(401) + _, stream := httphelpers.SSEHandler(nil) + defer stream.Close() + + streamManagerTestWithStreamHandler(t, handler, stream, noopTestCache{}, func(p streamManagerTestParams) { + p.streamManager.extendedRetryDelay = time.Millisecond + + readyCh := p.streamManager.Start() + // The cache read reports "nothing cached" by closing its channel, so Relay knows it + // cannot serve as soon as the key is rejected. It does not wait out initTimeout. + err := helpers.RequireValue(p.t, readyCh, time.Second, "timed out waiting for the failure report") + require.Error(p.t, err) + + p.mockLog.AssertMessageMatch(t, true, ldlog.Error, "no cached configuration is available") + }) +} + +func TestKeepsRunningWhenUnauthorizedButCacheHasConfiguration(t *testing.T) { + // The opposite case: cached configuration means Relay can serve, so it stays up past the + // grace period and keeps trying the key rather than exiting. + handler := httphelpers.HandlerWithStatus(401) + _, stream := httphelpers.SSEHandler(nil) + defer stream.Close() + + cache := &recordingCache{} + require.NoError(t, cache.SetAll(context.Background(), PutContent{ + Environments: map[config.EnvironmentID]envfactory.EnvironmentRep{testEnv1.EnvID: testEnv1}, + })) + + streamManagerTestWithStreamHandler(t, handler, stream, cache, func(p streamManagerTestParams) { + p.streamManager.extendedRetryDelay = time.Millisecond + + readyCh := p.streamManager.Start() + + // The cached environment reaches the handler, which is what makes Relay serviceable. + p.requireMessage() + p.requireReceivedAllMessage() + + // Well past initTimeout, nothing has been reported, so Relay is still running. + if !helpers.AssertNoMoreValues(t, readyCh, 1500*time.Millisecond, + "Relay reported a failure even though the cache supplied a configuration") { + t.FailNow() + } + p.mockLog.AssertMessageMatch(t, false, ldlog.Error, "no cached configuration is available") + p.mockLog.AssertMessageMatch(t, true, ldlog.Info, "loaded from persistent cache") + }) +} + +func TestIgnoreConnectionErrorsKeepsRunningWithNoConfiguration(t *testing.T) { + // ignoreConnectionErrors is documented as "go on trying to connect in the background while + // still allowing clients to connect to the Relay Proxy". With it set, a rejected key does + // not report a failure even though Relay has nothing to serve. + handler := httphelpers.HandlerWithStatus(401) + _, stream := httphelpers.SSEHandler(nil) + defer stream.Close() + + streamManagerTestWithStreamHandler(t, handler, stream, noopTestCache{}, func(p streamManagerTestParams) { + p.streamManager.extendedRetryDelay = time.Millisecond + p.streamManager.initTimeout = 50 * time.Millisecond + p.streamManager.ignoreConnectionErrors = true + + readyCh := p.streamManager.Start() + + if !helpers.AssertNoMoreValues(t, readyCh, 500*time.Millisecond, + "Relay reported a failure even though ignoreConnectionErrors is set") { + t.FailNow() + } + p.mockLog.AssertMessageMatch(t, false, ldlog.Error, "no cached configuration is available") + p.mockLog.AssertMessageMatch(t, true, ldlog.Error, "will keep retrying") + }) +} diff --git a/internal/autoconfig/stream_manager_test_base_test.go b/internal/autoconfig/stream_manager_test_base_test.go index c54e39394..a88b33275 100644 --- a/internal/autoconfig/stream_manager_test_base_test.go +++ b/internal/autoconfig/stream_manager_test_base_test.go @@ -308,6 +308,8 @@ func streamManagerTestWithStreamHandler( rpacProtocolVersion, mockLog.Loggers, cache, + time.Second, // initTimeout: bounds the wait for the cache read + false, // ignoreConnectionErrors ) defer p.streamManager.Close() diff --git a/internal/autoconfigcache/reload_restart_redis_test.go b/internal/autoconfigcache/reload_restart_redis_test.go index d89185e33..be6ecba4f 100644 --- a/internal/autoconfigcache/reload_restart_redis_test.go +++ b/internal/autoconfigcache/reload_restart_redis_test.go @@ -82,7 +82,10 @@ func TestConcurrentKeysCacheReloadSurvivesRestart(t *testing.T) { u, parseErr := url.Parse(streamURL) require.NoError(t, parseErr) return autoconfig.NewStreamManager(cacheConfig().AutoConfig.Key, u, handler, httpConfig, - time.Millisecond, restartProtocolV2, loggers, store) + time.Millisecond, restartProtocolV2, loggers, store, + time.Second, // initTimeout + false, // ignoreConnectionErrors + ) } // A multi-key environment (anchor + one extra SDK key, anchor + one extra mobile key) via the array diff --git a/relay/relay.go b/relay/relay.go index 64ea645b0..9a6d25b66 100644 --- a/relay/relay.go +++ b/relay/relay.go @@ -215,17 +215,25 @@ func newRelayInternal(c config.Config, options relayInternalOptions) (*Relay, er rpacProtocolVersion, loggers, autoConfigCache, + c.Main.InitTimeout.GetOrElse(config.DefaultInitTimeout), + c.Main.IgnoreConnectionErrors, ) autoConfigResult := r.autoConfigStream.Start() go func() { err := <-autoConfigResult if err != nil { - // This channel only emits a non-nil error if it's an unrecoverable error, in which case - // Relay should quit. The ExitOnError option doesn't affect this, because a failure of - // auto-config is more serious than any environment-specific failure; Relay can't possibly - // do anything useful without a configuration. The StreamManager has already logged the - // error by this point, so we just need to quit. + // This channel emits a non-nil error only when Relay has no configuration and no + // longer expects to get one, in which case Relay should quit. Relay cannot do + // anything useful without a configuration, so the ExitOnError option does not + // affect this; a failure of auto-config is more serious than any + // environment-specific failure. + // + // A rejected auto-configuration key no longer reaches this point on its own. The + // StreamManager keeps retrying, and reports a failure here only after a grace + // period during which no configuration arrived from either the stream or the + // persistent cache. With a cached configuration Relay stays running and serves + // from it. The StreamManager has already logged the reason by this point. os.Exit(1) } }() From 789fb2e51dc666b260d022ae955c50e6385a8239 Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Wed, 9 Sep 2026 11:23:03 -0400 Subject: [PATCH 2/4] test: Pin that a rejected key engages the extended delays Applying the big segment review's method to this change found the same kind of gap. Removing the profile activation from the error handler left every auto-configuration test green: they proved the stream keeps retrying, but nothing proved it retries slowly, which is the reason for the change. eventsource computes the delay after applying the profile the error handler returns, and logs it, so the first log line after a rejected key already reflects the extended delays. The test reads that line and returns, so it never waits the delay out. Verified that the test fails when the activation is removed, and that removing the notification the error handler sends, the option guard, and the cache result each fail a different test. --- .../autoconfig/stream_manager_errors_test.go | 56 +++++++++++++++++++ 1 file changed, 56 insertions(+) diff --git a/internal/autoconfig/stream_manager_errors_test.go b/internal/autoconfig/stream_manager_errors_test.go index 8c15a70bb..74bc1ee58 100644 --- a/internal/autoconfig/stream_manager_errors_test.go +++ b/internal/autoconfig/stream_manager_errors_test.go @@ -4,6 +4,8 @@ import ( "context" "fmt" "net/http" + "strconv" + "strings" "testing" "time" @@ -11,6 +13,7 @@ import ( "github.com/stretchr/testify/require" "github.com/launchdarkly/go-sdk-common/v3/ldlog" + "github.com/launchdarkly/go-sdk-common/v3/ldlogtest" helpers "github.com/launchdarkly/go-test-helpers/v3" "github.com/launchdarkly/go-test-helpers/v3/httphelpers" @@ -450,3 +453,56 @@ func TestIgnoreConnectionErrorsKeepsRunningWithNoConfiguration(t *testing.T) { p.mockLog.AssertMessageMatch(t, true, ldlog.Error, "will keep retrying") }) } + +// firstRetryDelay returns the delay from eventsource's first "retrying in N secs" line, which +// it logs through the loggers Relay supplies. +func firstRetryDelay(mockLog *ldlogtest.MockLog) (time.Duration, bool) { + for _, line := range mockLog.GetOutput(ldlog.Info) { + _, after, found := strings.Cut(line, "retrying in ") + if !found { + continue + } + secs, err := strconv.ParseFloat(strings.TrimSuffix(strings.TrimSpace(after), " secs"), 64) + if err != nil { + continue + } + return time.Duration(secs * float64(time.Second)), true + } + return 0, false +} + +func TestUnauthorizedEngagesTheExtendedDelays(t *testing.T) { + // Keeping the stream alive is only half the point. It must also retry slowly, so that a + // fleet of Relay Proxy instances does not hammer a service that is rejecting every + // request. Without this, removing the profile activation leaves every other test green. + // + // eventsource computes the delay after applying the profile the error handler returns, and + // logs it, so the first line already reflects the extended delays. The delay is never + // waited out: the test reads the log line and returns. + const extendedDelay = 10 * time.Minute + + handler := httphelpers.HandlerWithStatus(401) + _, stream := httphelpers.SSEHandler(nil) + defer stream.Close() + + streamManagerTestWithStreamHandler(t, handler, stream, noopTestCache{}, func(p streamManagerTestParams) { + p.streamManager.extendedRetryDelay = extendedDelay + // Keep Relay alive, so it does not report the failure and abandon the stream before + // the log line appears. + p.streamManager.ignoreConnectionErrors = true + + p.streamManager.Start() + + var delay time.Duration + require.Eventually(t, func() bool { + d, ok := firstRetryDelay(p.mockLog) + delay = d + return ok + }, time.Second, 5*time.Millisecond, "expected eventsource to log a retry delay") + + // Jitter removes up to half the delay, so the floor is half the base. The normal + // ceiling is 30s, so any delay above it can only have come from the extended profile. + assert.GreaterOrEqual(t, delay, extendedDelay/2, "the delay must come from the extended profile") + assert.LessOrEqual(t, delay, extendedDelay) + }) +} From 072e62be35094872bdf6a83d8ba94c3fb8309252 Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Tue, 15 Sep 2026 10:26:06 -0400 Subject: [PATCH 3/4] fix: Address the review findings on the retry behavior A multi-agent review found that the change did not do what it claimed for the common configuration, and that it had widened the set of failures that stop the process. Each finding below now has a test that fails without the fix. Only a rejected credential can stop Relay. The classifier now reports the retry class and whether the credential was rejected as two separate answers, because they decide different things: the class picks how long to wait, and only a rejection means Relay can never succeed. Previously every failure in the slow-retry class reached the give-up path, so a 404 from a misconfigured stream URI, or a certificate problem, exited a Relay with no cached configuration where before it retried indefinitely. No transport failure is unexpected any more. Certificate failures join the rest on the short delays, following the direction the retry specification is taking. That left nothing for the transport classifier to decide, so it is gone and the policy is stated in the package documentation instead. Big segment synchronization picks up the same change. A store Relay cannot read is not an empty store. The cache read now reports "unavailable" separately from "empty", so a failover or a timeout leaves Relay running instead of concluding it has nothing to serve and exiting while the store holds a usable configuration. An entry with no environments does count as empty: without environments there is no credential to accept and no data to answer with. Shutdown interrupts a pending backoff wait. The stream now runs on a cancellable context, which is the only thing the retry loop selects on besides its delay timer. Without it the connection attempt outlived Close by as long as the delay, up to an hour, and then presented the rejected credential again. Giving up cancels it too. initTimeout bounds the cache read rather than racing it in the wait loop. That removes the timer and one of the select arms, and makes a non-positive value fall back to the default rather than mean "give up at once", which had discarded a configuration the store was about to deliver. Log lines forwarded from the SSE library are sanitised. Their text embeds the response body of a rejected request, which comes from whatever answered it, so newlines let it forge log entries and an error page could arrive at any length. --- docs/proxy-mode.md | 6 +- internal/autoconfig/errors_and_messages.go | 2 + internal/autoconfig/stream_logger.go | 49 +++++ internal/autoconfig/stream_logger_test.go | 61 ++++++ internal/autoconfig/stream_manager.go | 176 +++++++++++------- .../autoconfig/stream_manager_giveup_test.go | 142 ++++++++++++++ .../stream_manager_known_limits_test.go | 161 ++++++++++++++++ .../stream_manager_retry_profile_test.go | 65 +++++++ .../stream_manager_shutdown_test.go | 128 +++++++++++++ .../stream_manager_stop_policy_test.go | 98 ++++++++++ relay/relay.go | 11 +- 11 files changed, 826 insertions(+), 73 deletions(-) create mode 100644 internal/autoconfig/stream_logger.go create mode 100644 internal/autoconfig/stream_logger_test.go create mode 100644 internal/autoconfig/stream_manager_giveup_test.go create mode 100644 internal/autoconfig/stream_manager_known_limits_test.go create mode 100644 internal/autoconfig/stream_manager_retry_profile_test.go create mode 100644 internal/autoconfig/stream_manager_shutdown_test.go create mode 100644 internal/autoconfig/stream_manager_stop_policy_test.go diff --git a/docs/proxy-mode.md b/docs/proxy-mode.md index af77226d5..62267cea0 100644 --- a/docs/proxy-mode.md +++ b/docs/proxy-mode.md @@ -60,10 +60,14 @@ What happens to SDK requests in the meantime depends on whether the Relay Proxy * If you configure a [persistent store](./persistent-storage.md) and it holds configuration data from a previous run, the Relay Proxy loads that data and serves those environments while it keeps retrying. It logs `AutoConfig loaded from persistent cache`. -* If the Relay Proxy has no configuration from any source, it cannot serve any request, because it does not know what its environments are. It shuts down, so that whatever supervises the process reports the problem. This happens as soon as the Relay Proxy has both been rejected and finished reading the persistent store, which is usually immediate. If the store does not answer, the Relay Proxy waits no longer than [`initTimeout`](./configuration.md#file-section-main) before shutting down. +* If the Relay Proxy has no configuration from any source, it cannot serve any request, because it does not know what its environments are. It shuts down, so that whatever supervises the process reports the problem. This happens on the first rejection, as soon as the Relay Proxy knows the persistent store holds nothing for it. To keep the Relay Proxy running and retrying in this case instead, set [`ignoreConnectionErrors`](./configuration.md#file-section-main) to `true`. Note that until the key becomes valid, the Relay Proxy still answers every request with a `503` error, because it has no environments to serve. + A store the Relay Proxy cannot read is not the same as an empty store. If the read fails or does not finish within [`initTimeout`](./configuration.md#file-section-main), the Relay Proxy keeps running and retrying rather than shutting down, because the store may hold a configuration it could serve. + +Only a `401` or `403` causes this. Any other failure to reach LaunchDarkly, including a `404`, a `5xx`, a network error, or a certificate problem, leaves the Relay Proxy running and retrying as it did before, on the schedule described above. + Earlier versions of the Relay Proxy shut down immediately when LaunchDarkly rejected the auto-configuration key, even when a persistent store held usable configuration data. ### Relay Proxy receives a request with invalid credentials diff --git a/internal/autoconfig/errors_and_messages.go b/internal/autoconfig/errors_and_messages.go index 58e1a11a7..18cf358ad 100644 --- a/internal/autoconfig/errors_and_messages.go +++ b/internal/autoconfig/errors_and_messages.go @@ -8,6 +8,8 @@ const ( logMsgBadKey = "Invalid auto-configuration key; cannot get environments" logMsgBadKeyWillRetry = "Invalid auto-configuration key; will keep retrying in case it becomes valid" logMsgExtendedBackoff = "Classified failure as UNEXPECTED; engaging extended backoff." + logMsgCacheReadFailed = "AutoConfig cache read failed (will rely on stream): %v" + logMsgCacheReadTimeout = "AutoConfig cache read did not finish within %s (will rely on stream)" logMsgNoConfigGaveUp = "Cannot get environments and no cached configuration is available; " + "Relay cannot serve requests. Set ignoreConnectionErrors to keep Relay running and retrying" logMsgDeliberateReconnect = "Will restart auto-configuration stream to get new data due to a policy change" diff --git a/internal/autoconfig/stream_logger.go b/internal/autoconfig/stream_logger.go new file mode 100644 index 000000000..13289c9c1 --- /dev/null +++ b/internal/autoconfig/stream_logger.go @@ -0,0 +1,49 @@ +package autoconfig + +import ( + "fmt" + "strings" + "unicode" + + "github.com/launchdarkly/go-sdk-common/v3/ldlog" +) + +// maxStreamLogLineLength caps a line forwarded from the SSE library. The text can embed an +// error response body of any size, and a log line is not the place to reproduce one. +const maxStreamLogLineLength = 200 + +// streamLogger forwards the SSE library's log output to Relay's logger after making it safe to +// write to a log. +// +// The library reports a failed connection as "Connection failed (%s), retrying in ... secs", +// where the error text for a rejected request embeds the response body verbatim. That body +// comes from whatever answered the request, which may be an intermediary rather than +// LaunchDarkly. Passing it through unaltered lets it end a line and start another, so it can +// forge log entries; and an error page can be arbitrarily long. +type streamLogger struct { + dest ldlog.BaseLogger +} + +func (l streamLogger) Println(values ...interface{}) { + l.dest.Println(sanitizeStreamLogLine(fmt.Sprint(values...))) +} + +func (l streamLogger) Printf(format string, values ...interface{}) { + l.dest.Println(sanitizeStreamLogLine(fmt.Sprintf(format, values...))) +} + +// sanitizeStreamLogLine collapses every control character, so the text cannot span lines, and +// truncates it to a length a log can reasonably carry. +func sanitizeStreamLogLine(s string) string { + s = strings.Map(func(r rune) rune { + if r == '\t' || !unicode.IsControl(r) { + return r + } + return ' ' + }, s) + s = strings.TrimSpace(s) + if len(s) > maxStreamLogLineLength { + return s[:maxStreamLogLineLength] + "... (truncated)" + } + return s +} diff --git a/internal/autoconfig/stream_logger_test.go b/internal/autoconfig/stream_logger_test.go new file mode 100644 index 000000000..2dc404e57 --- /dev/null +++ b/internal/autoconfig/stream_logger_test.go @@ -0,0 +1,61 @@ +package autoconfig + +import ( + "net/http" + "strings" + "testing" + "time" + + "github.com/launchdarkly/go-sdk-common/v3/ldlog" + "github.com/launchdarkly/go-test-helpers/v3/httphelpers" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestSanitizeStreamLogLine(t *testing.T) { + assert.Equal(t, "plain text", sanitizeStreamLogLine("plain text")) + + // A newline would let upstream text end the line and start a forged one. + assert.Equal(t, "error 401: denied FAKE Error: forged line", + sanitizeStreamLogLine("error 401: denied\nFAKE Error: forged line")) + assert.Equal(t, "a b c", sanitizeStreamLogLine("a\rb\x00c")) + + // Tabs are legible in a log and carry no line-ending risk. + assert.Equal(t, "a\tb", sanitizeStreamLogLine("a\tb")) + + long := sanitizeStreamLogLine(strings.Repeat("x", maxStreamLogLineLength+50)) + assert.Len(t, long, maxStreamLogLineLength+len("... (truncated)")) + assert.True(t, strings.HasSuffix(long, "... (truncated)")) +} + +// The response body of a rejected request reaches the log through the SSE library's error text. +// It comes from whatever answered the request, so it must not be able to forge a log line. +func TestResponseBodyCannotForgeALogLine(t *testing.T) { + const canary = "CANARY-BODY" + handler := func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusUnauthorized) + _, _ = w.Write([]byte(canary + "\nFAKE Error: forged log line\n" + strings.Repeat("padding ", 60))) + } + _, stream := httphelpers.SSEHandler(nil) + defer stream.Close() + + streamManagerTestWithStreamHandler(t, http.HandlerFunc(handler), stream, noopTestCache{}, + func(p streamManagerTestParams) { + p.streamManager.extendedRetryDelay = time.Millisecond + p.streamManager.ignoreConnectionErrors = true // stay alive long enough to log + p.streamManager.Start() + + require.Eventually(t, func() bool { + return hasAnyMessage(p.mockLog, ldlog.Info, "Connection failed") + }, 2*time.Second, 10*time.Millisecond, "expected the library's retry line") + + for _, line := range p.mockLog.GetOutput(ldlog.Info) { + if !strings.Contains(line, "Connection failed") { + continue + } + assert.NotContains(t, line, "\n", "the body must not be able to end the line") + assert.LessOrEqual(t, len(line), maxStreamLogLineLength+len("... (truncated)")+40, + "an unbounded body must not reach the log in full") + } + }) +} diff --git a/internal/autoconfig/stream_manager.go b/internal/autoconfig/stream_manager.go index 70ec193a8..dfe1f282c 100644 --- a/internal/autoconfig/stream_manager.go +++ b/internal/autoconfig/stream_manager.go @@ -100,9 +100,15 @@ type StreamManager struct { done chan struct{} // closed when the subscribe goroutine exits closeOnce sync.Once + // streamCtx bounds the lifetime of the SSE connection, including any backoff wait between + // attempts. Cancelling it is the only way to interrupt that wait: eventsource's retry loop + // selects on the request's context and the delay timer, and nothing else. + streamCtx context.Context + streamCancel context.CancelFunc + // cacheCh receives the result of the async cache read started by Start(). // It is consumed by consumeStream and nilled out after use. - cacheCh <-chan *PutContent + cacheCh <-chan cacheReadResult cacheCancel context.CancelFunc envReceiver *MessageReceiver[envfactory.EnvironmentRep] @@ -137,8 +143,11 @@ func NewStreamManager( protocolVersionParam: []string{strconv.Itoa(protocolVersion)}, }.Encode() } + streamCtx, streamCancel := context.WithCancel(context.Background()) s := &StreamManager{ key: key, + streamCtx: streamCtx, + streamCancel: streamCancel, uri: streamURI, handler: handler, cache: cache, @@ -174,6 +183,17 @@ func NewStreamManager( return s } +// cacheReadTimeout is how long the cache read may take before Relay treats the cache as +// unavailable. It is the initTimeout configuration option, with non-positive values replaced by +// the default: zero means "do not block" for the SDK client elsewhere in Relay, and applying +// that reading here would discard a usable cached configuration before the store could answer. +func (s *StreamManager) cacheReadTimeout() time.Duration { + if s.initTimeout <= 0 { + return config.DefaultInitTimeout + } + return s.initTimeout +} + // Start causes the StreamManager to start trying to connect to the auto-config stream. The returned channel // receives nil for a successful connection, or an error if it has permanently failed. // @@ -182,19 +202,30 @@ func NewStreamManager( // the cache read is cancelled and its result discarded. func (s *StreamManager) Start() <-chan error { // Start the cache read concurrently with the stream connection. - cacheCtx, cacheCancel := context.WithCancel(context.Background()) - cacheCh := make(chan *PutContent, 1) + // + // The read is bounded by initTimeout. Neither cache store sets a deadline of its own, so + // without this a wedged store would leave Relay unable to decide whether it has anything + // to serve. Bounding it here rather than in the wait loop means the loop has one less + // thing to race, and the timeout produces the same closed channel as an empty cache. + cacheCtx, cacheCancel := context.WithTimeout(context.Background(), s.cacheReadTimeout()) + cacheCh := make(chan cacheReadResult, 1) go func() { defer close(cacheCh) content, err := s.cache.GetAll(cacheCtx) if err != nil { - if cacheCtx.Err() == nil { - s.loggers.Warnf("AutoConfig cache read failed (will rely on stream): %v", err) + switch { + case errors.Is(err, context.DeadlineExceeded): + s.loggers.Warnf(logMsgCacheReadTimeout, s.cacheReadTimeout()) + case cacheCtx.Err() == nil: + s.loggers.Warnf(logMsgCacheReadFailed, err) + default: + return // cancelled because Relay no longer needs the result } + cacheCh <- cacheReadResult{unavailable: true} return } if content != nil { - cacheCh <- content + cacheCh <- cacheReadResult{content: content} } }() s.cacheCh = cacheCh @@ -215,6 +246,9 @@ func (s *StreamManager) Start() <-chan error { func (s *StreamManager) Close() { s.closeOnce.Do(func() { close(s.halt) + // halt is only observed when the next attempt fails, so on its own it cannot end a + // backoff wait that is already pending. Cancelling the stream context does. + s.streamCancel() s.updateStatus(interfaces.DataSourceStateOff, interfaces.DataSourceErrorInfo{}) }) if s.done != nil { @@ -223,6 +257,15 @@ func (s *StreamManager) Close() { _ = s.cache.Close() } +// cacheReadResult carries the outcome of the startup cache read. A closed channel with no +// value means the cache is reachable and empty. A value with unavailable set means the read +// failed or timed out, which is not the same thing: the store may well hold a usable +// configuration, so Relay must not conclude from it that there is nothing to serve. +type cacheReadResult struct { + content *PutContent + unavailable bool +} + type streamResult struct { stream *es.Stream err error @@ -258,10 +301,10 @@ func (s *StreamManager) subscribe(readyCh chan<- error) { es.RetryProfileJitter(streamJitterRatio), ) - // authFailureCh tells the wait loop below that the service rejected the key. The send - // never blocks, and one notification is enough. - authFailureCh := make(chan struct{}, 1) - errorHandler := s.newStreamErrorHandler(authFailureCh, extendedProfile) + // keyRejectedCh tells the wait loop below that the service rejected the credential. The + // send never blocks, and one notification is enough. + keyRejectedCh := make(chan struct{}, 1) + errorHandler := s.newStreamErrorHandler(keyRejectedCh, extendedProfile) rpacEndpoint, err := url.JoinPath(s.uri.String(), autoConfigStreamPath) if err != nil { @@ -270,7 +313,7 @@ func (s *StreamManager) subscribe(readyCh chan<- error) { return } - req, _ := http.NewRequest("GET", rpacEndpoint, nil) + req, _ := http.NewRequestWithContext(s.streamCtx, "GET", rpacEndpoint, nil) req.Header.Set("Authorization", string(s.key)) s.loggers.Infof(logMsgStreamConnecting, rpacEndpoint) @@ -290,7 +333,7 @@ func (s *StreamManager) subscribe(readyCh chan<- error) { es.StreamOptionRetryResetInterval(streamRetryResetInterval), es.StreamOptionErrorHandler(errorHandler), es.StreamOptionCanRetryFirstConnection(-1), - es.StreamOptionLogger(s.loggers.ForLevel(ldlog.Info)), + es.StreamOptionLogger(streamLogger{dest: s.loggers.ForLevel(ldlog.Info)}), ) streamCh <- streamResult{stream, err} }() @@ -302,41 +345,41 @@ func (s *StreamManager) subscribe(readyCh chan<- error) { // them before the stream connects, so these three facts decide whether waiting is still // worthwhile. haveConfiguration := false // the cache supplied a configuration - cacheReported := false // the cache read finished, with or without data - authFailed := false // the service rejected the key + cacheEmpty := false // the cache is reachable and holds nothing + keyRejected := false // the service rejected the credential // shouldGiveUp reports whether Relay has established that it cannot serve. That needs a - // rejected key and a finished cache read that produced nothing. Any other failure leaves - // Relay waiting and retrying, which is what it did before this change. + // rejected credential and a cache that is reachable and empty. A cache Relay could not + // read is deliberately not enough: the store may hold a usable configuration, and exiting + // on a failover or a DNS blip would throw it away. Any other failure leaves Relay waiting + // and retrying, which is what it did before this change. shouldGiveUp := func() bool { - return authFailed && cacheReported && !haveConfiguration && !s.ignoreConnectionErrors + return keyRejected && cacheEmpty && !haveConfiguration && !s.ignoreConnectionErrors } giveUp := func() { s.loggers.Error(logMsgNoConfigGaveUp) signalReady(errors.New("invalid auto-configuration key")) + // Stop the connection attempt as well. Without this the abandoned goroutine keeps + // presenting a credential the service has already rejected, for as long as the + // extended delays run. + s.streamCancel() s.abandonStreamGoroutine(streamCh) } - // initTimeoutCh stays nil, and so blocks forever in the select, until the service rejects - // the key. It bounds the wait for the cache read, which has no deadline of its own: - // neither cache store sets one, and both rely on their driver's network timeouts. - var initTimer *time.Timer - var initTimeoutCh <-chan time.Time - defer func() { - if initTimer != nil { - initTimer.Stop() - } - }() - var stream *es.Stream for stream == nil { select { - case content, ok := <-s.cacheCh: - cacheReported = true - if ok && content != nil { - s.applyCachedContent(content) + case result, ok := <-s.cacheCh: + switch { + case result.content != nil && len(result.content.Environments) > 0: + s.applyCachedContent(result.content) haveConfiguration = true + case !ok || !result.unavailable: + // Reachable but with nothing Relay can serve. An entry holding only filters + // counts as nothing: without environments there is no credential to accept and + // no flag data to answer with. + cacheEmpty = true } if s.cacheCancel != nil { s.cacheCancel() @@ -348,25 +391,12 @@ func (s *StreamManager) subscribe(readyCh chan<- error) { return } - case <-authFailureCh: - authFailed = true + case <-keyRejectedCh: + keyRejected = true if shouldGiveUp() { giveUp() return } - if initTimer == nil && !s.ignoreConnectionErrors { - initTimer = time.NewTimer(s.initTimeout) - initTimeoutCh = initTimer.C - } - - case <-initTimeoutCh: - initTimeoutCh = nil - if !haveConfiguration { - // The cache never reported. Waiting longer cannot help, because the key is - // rejected and there is nothing to serve. - giveUp() - return - } case result := <-streamCh: if result.err != nil { @@ -404,9 +434,11 @@ func (s *StreamManager) consumeStream(stream *es.Stream) { for { select { - case content, ok := <-s.cacheCh: - if ok && content != nil { - s.applyCachedContent(content) + case result, ok := <-s.cacheCh: + // The stream is already connected here, so a late cache result is only useful if + // it carries data; an unreadable cache changes nothing. + if ok && result.content != nil { + s.applyCachedContent(result.content) } if s.cacheCancel != nil { s.cacheCancel() @@ -611,7 +643,7 @@ func (s *StreamManager) dispatchFilterAction(id config.FilterID, rep envfactory. // stream to the longer delays and it keeps trying. The handler reports such a failure on // authFailureCh, because only the caller can decide whether Relay is able to serve meanwhile. func (s *StreamManager) newStreamErrorHandler( - authFailureCh chan<- struct{}, + keyRejectedCh chan<- struct{}, extendedProfile *es.RetryProfile, ) func(error) es.StreamErrorHandlerResult { // loggedExtended keeps the notice about the longer delays to once per subscribe cycle. @@ -627,7 +659,7 @@ func (s *StreamManager) newStreamErrorHandler( default: } - class := s.classifyAndLogStreamError(err) + class, keyRejected := s.classifyAndLogStreamError(err) result := es.StreamErrorHandlerResult{CloseNow: false} if class == retry.Unexpected { @@ -636,8 +668,14 @@ func (s *StreamManager) newStreamErrorHandler( loggedExtended = true } result.ActivateProfile = extendedProfile + } + + // Only a rejected credential can make Relay stop. Every other failure, however slowly + // it retries, leaves Relay running: a 404 from a misconfigured stream URI or a + // certificate problem is not a reason to take a whole fleet down. + if keyRejected { select { - case authFailureCh <- struct{}{}: + case keyRejectedCh <- struct{}{}: default: } } @@ -645,31 +683,35 @@ func (s *StreamManager) newStreamErrorHandler( } } -// classifyAndLogStreamError sorts a stream failure into a retry class and logs it. A failure -// that is unlikely to correct itself soon is worth an error, because it nearly always means a -// real configuration problem, even though the stream recovers on its own once it is fixed. -func (s *StreamManager) classifyAndLogStreamError(err error) retry.FailureClass { +// classifyAndLogStreamError sorts a stream failure into a retry class, logs it, and reports +// whether the service rejected the credential. +// +// The two results answer different questions. The class decides how long to wait before the +// next attempt. Only a rejected credential decides whether Relay can ever succeed, so only it +// can lead to Relay stopping. +// +// A failure that is unlikely to correct itself soon is worth an error, because it nearly +// always means a real configuration problem, even though the stream recovers on its own once +// that problem is fixed. +func (s *StreamManager) classifyAndLogStreamError(err error) (class retry.FailureClass, keyRejected bool) { var se es.SubscriptionError if !errors.As(err, &se) { - class := retry.ClassifyTransportError(err) - if class == retry.Unexpected { - s.loggers.Errorf(logMsgStreamOtherError, err) - } else { - s.loggers.Warnf(logMsgStreamOtherError, err) - } - return class + // No transport-level failure is unexpected, so these keep the short delays. + s.loggers.Warnf(logMsgStreamOtherError, err) + return retry.Normal, false } - class := retry.ClassifyHTTPStatus(se.Code) + class = retry.ClassifyHTTPStatus(se.Code) + keyRejected = se.Code == http.StatusUnauthorized || se.Code == http.StatusForbidden switch { - case se.Code == 401 || se.Code == 403: + case keyRejected: s.loggers.Error(logMsgBadKeyWillRetry) case class == retry.Unexpected: s.loggers.Errorf(logMsgStreamHTTPError, se.Code) default: s.loggers.Warnf(logMsgStreamHTTPError, se.Code) } - return class + return class, keyRejected } // abandonStreamGoroutine leaves the SSE connection attempt behind. That goroutine may still diff --git a/internal/autoconfig/stream_manager_giveup_test.go b/internal/autoconfig/stream_manager_giveup_test.go new file mode 100644 index 000000000..77ceed32f --- /dev/null +++ b/internal/autoconfig/stream_manager_giveup_test.go @@ -0,0 +1,142 @@ +package autoconfig + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/launchdarkly/go-sdk-common/v3/ldlog" + helpers "github.com/launchdarkly/go-test-helpers/v3" + "github.com/launchdarkly/go-test-helpers/v3/httphelpers" + "github.com/stretchr/testify/require" + + "github.com/launchdarkly/ld-relay/v8/config" + "github.com/launchdarkly/ld-relay/v8/internal/envfactory" +) + +// slowCache answers GetAll with usable content, but only after a delay. +type slowCache struct { + delay time.Duration + content *PutContent +} + +func (c *slowCache) GetAll(ctx context.Context) (*PutContent, error) { + select { + case <-time.After(c.delay): + return c.content, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} +func (c *slowCache) SetAll(context.Context, PutContent) error { return nil } +func (c *slowCache) Upsert(context.Context, CacheKind, string, interface{}) error { return nil } +func (c *slowCache) Delete(context.Context, CacheKind, string) error { return nil } +func (c *slowCache) Close() error { return nil } + +// erroringCache fails the read, the way a store does during a failover, even though the store +// still holds a usable configuration. +type erroringCache struct{} + +func (erroringCache) GetAll(context.Context) (*PutContent, error) { + return nil, errors.New("connection refused: store is failing over") +} +func (erroringCache) SetAll(context.Context, PutContent) error { return nil } +func (erroringCache) Upsert(context.Context, CacheKind, string, interface{}) error { return nil } +func (erroringCache) Delete(context.Context, CacheKind, string) error { return nil } +func (erroringCache) Close() error { return nil } + +func oneEnvironmentCacheContent() *PutContent { + return &PutContent{ + Environments: map[config.EnvironmentID]envfactory.EnvironmentRep{testEnv1.EnvID: testEnv1}, + } +} + +// rejectingStreamTest runs a StreamManager against a stream that rejects every request, with the +// given cache, and returns the readiness channel. +func rejectingStreamTest( + t *testing.T, + cache Cache, + configure func(p streamManagerTestParams), + action func(p streamManagerTestParams, readyCh <-chan error), +) { + t.Helper() + handler := httphelpers.HandlerWithStatus(401) + _, stream := httphelpers.SSEHandler(nil) + defer stream.Close() + + streamManagerTestWithStreamHandler(t, handler, stream, cache, func(p streamManagerTestParams) { + p.streamManager.extendedRetryDelay = time.Millisecond + if configure != nil { + configure(p) + } + action(p, p.streamManager.Start()) + }) +} + +// A non-positive initTimeout must not mean "give up at once". Zero means "do not block on init" +// everywhere else in Relay's configuration surface, and applying that reading here discarded a +// cached configuration the store was about to deliver -- the give-up cancelled the in-flight +// read on its way out. +func TestNonPositiveInitTimeoutStillWaitsForTheCacheRead(t *testing.T) { + for _, initTimeout := range []time.Duration{0, -1 * time.Second} { + t.Run(initTimeout.String(), func(t *testing.T) { + cache := &slowCache{delay: 300 * time.Millisecond, content: oneEnvironmentCacheContent()} + + rejectingStreamTest(t, cache, + func(p streamManagerTestParams) { p.streamManager.initTimeout = initTimeout }, + func(p streamManagerTestParams, readyCh <-chan error) { + // The cached environment still arrives and Relay keeps serving it. + p.requireMessage() + p.requireReceivedAllMessage() + if !helpers.AssertNoMoreValues(t, readyCh, 300*time.Millisecond, + "Relay gave up despite a usable cached configuration") { + t.FailNow() + } + }) + }) + } +} + +// A cache Relay could not read is not an empty cache. A failover or a DNS blip at startup must +// not make Relay conclude it has nothing to serve, because the store may hold a perfectly good +// configuration and there is no second read. +func TestUnreadableCacheDoesNotCountAsEmpty(t *testing.T) { + rejectingStreamTest(t, erroringCache{}, nil, + func(p streamManagerTestParams, readyCh <-chan error) { + if !helpers.AssertNoMoreValues(t, readyCh, 700*time.Millisecond, + "Relay gave up on a cache read failure, which is not the same as an empty cache") { + t.FailNow() + } + p.mockLog.AssertMessageMatch(t, true, ldlog.Warn, "cache read failed") + p.mockLog.AssertMessageMatch(t, false, ldlog.Error, "no cached configuration is available") + }) +} + +// A reachable, genuinely empty cache is still grounds for giving up -- the other half of the +// distinction above. +func TestEmptyCacheStillGivesUp(t *testing.T) { + rejectingStreamTest(t, noopTestCache{}, nil, + func(p streamManagerTestParams, readyCh <-chan error) { + err := helpers.RequireValue(t, readyCh, time.Second, + "expected Relay to give up on a rejected credential with an empty cache") + require.Error(t, err) + p.mockLog.AssertMessageMatch(t, true, ldlog.Error, "no cached configuration is available") + }) +} + +// An entry holding only filters is nothing Relay can serve: no environments means no credential +// to accept and no flag data to answer with. It must not count as a configuration. +func TestFiltersOnlyCacheIsNotAConfiguration(t *testing.T) { + cache := &slowCache{content: &PutContent{ + Filters: map[config.FilterID]envfactory.FilterRep{"filter-1": {}}, + }} + + rejectingStreamTest(t, cache, nil, + func(p streamManagerTestParams, readyCh <-chan error) { + err := helpers.RequireValue(t, readyCh, time.Second, + "expected Relay to give up: a filters-only entry leaves it with zero environments") + require.Error(t, err) + p.mockLog.AssertMessageMatch(t, true, ldlog.Error, "no cached configuration is available") + }) +} diff --git a/internal/autoconfig/stream_manager_known_limits_test.go b/internal/autoconfig/stream_manager_known_limits_test.go new file mode 100644 index 000000000..e5c6310ed --- /dev/null +++ b/internal/autoconfig/stream_manager_known_limits_test.go @@ -0,0 +1,161 @@ +package autoconfig + +// Tests in this file pin behavior Relay currently has and that we have decided not to change +// yet. They are not proofs of pending bugs: each one asserts today's behavior so that a future +// fix shows up as a failing test somebody has to update deliberately, rather than a silent +// change. The comment on each test says what the limitation is and why it is tolerated. + +import ( + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "strconv" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/launchdarkly/go-sdk-common/v3/ldlog" + "github.com/launchdarkly/go-sdk-common/v3/ldlogtest" + helpers "github.com/launchdarkly/go-test-helpers/v3" + "github.com/launchdarkly/go-test-helpers/v3/httphelpers" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/launchdarkly/ld-relay/v8/config" + "github.com/launchdarkly/ld-relay/v8/internal/httpconfig" +) + +// TestKnownLimitGiveUpUnreachableOnceTheStreamHasConnected shows the give-up predicate is only ever +// evaluated in the pre-connection wait loop (stream_manager.go:318-372). Once +// SubscribeWithRequestAndOptions has returned a stream, subscribe calls signalReady(nil) and +// enters consumeStream, and the authFailureCh send at stream_manager.go:625-628 has no reader +// forever after. A stream that connects with HTTP 200 but delivers no PUT therefore leaves +// Relay reporting a successful init with zero environments, and no later rejection of the key - +// however permanent - can make it report a failure. +func TestKnownLimitGiveUpUnreachableOnceTheStreamHasConnected(t *testing.T) { + sseHandler, sseControl := httphelpers.SSEHandler(nil) // 200, no events + defer sseControl.Close() + handler := httphelpers.SequentialHandler(sseHandler, httphelpers.HandlerWithStatus(401)) + + streamManagerTestWithStreamHandler(t, handler, sseControl, noopTestCache{}, func(p streamManagerTestParams) { + p.streamManager.extendedRetryDelay = time.Millisecond + p.streamManager.initTimeout = 100 * time.Millisecond + require.False(t, p.streamManager.ignoreConnectionErrors) + + readyCh := p.streamManager.Start() + + // Init "succeeded" on the bare HTTP 200, with no configuration of any kind. + err := helpers.RequireValue(t, readyCh, 2*time.Second, "timed out waiting for the ready signal") + require.NoError(t, err) + p.mockLog.AssertMessageMatch(t, false, ldlog.Info, "Received configuration for") + + // Now the key is rejected on every reconnect, forever. + sseControl.EndAll() + + require.Eventually(t, func() bool { + return len(p.mockLog.GetOutput(ldlog.Error)) > 0 && + containsMatch(p.mockLog.GetOutput(ldlog.Error), "will keep retrying") + }, 2*time.Second, 10*time.Millisecond, "expected the rejection to be logged") + + // No give-up, ever, even though Relay has nothing to serve and ignoreConnectionErrors is false. + if !helpers.AssertNoMoreValues(t, readyCh, time.Second, "Relay reported a failure") { + t.FailNow() + } + p.mockLog.AssertMessageMatch(t, false, ldlog.Error, "no cached configuration is available") + }) +} + +func containsMatch(lines []string, sub string) bool { + for _, l := range lines { + if len(l) >= len(sub) { + for i := 0; i+len(sub) <= len(l); i++ { + if l[i:i+len(sub)] == sub { + return true + } + } + } + } + return false +} + +// reconnectDelays returns every delay eventsource logged, from both the first-connection loop +// ("retrying in N secs") and the post-connection loop ("Reconnecting in N secs"). +func reconnectDelays(mockLog *ldlogtest.MockLog) []time.Duration { + var out []time.Duration + for _, line := range mockLog.GetOutput(ldlog.Info) { + for _, marker := range []string{"retrying in ", "Reconnecting in "} { + _, after, found := strings.Cut(line, marker) + if !found { + continue + } + secs, err := strconv.ParseFloat(strings.TrimSuffix(strings.TrimSpace(after), " secs"), 64) + if err == nil { + out = append(out, time.Duration(secs*float64(time.Second))) + } + } + } + return out +} + +// TestKnownLimitServerRetryHintDefeatsTheExtendedProfile: eventsource's baseDelayOverride (set by an +// SSE "retry:" field) replaces the active profile's base delay, including the extended +// profile's, and is never cleared (retry_delay.go NextRetryDelay: "if +// activeState.baseDelayOverride != nil { effectiveBase = *activeState.baseDelayOverride }"). +// So once the stream has seen a retry hint, engaging the extended profile does not slow +// retries down to the intended 5 min: Relay keeps hammering the rejecting service on the +// hinted base delay instead. +func TestKnownLimitServerRetryHintDefeatsTheExtendedProfile(t *testing.T) { + const hintMillis = 10 + var requestCount int64 + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + n := atomic.AddInt64(&requestCount, 1) + if n > 1 { + w.WriteHeader(401) // the key is rejected from now on + return + } + // One good connection that carries a server-directed retry hint, then it drops. + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(200) + fmt.Fprintf(w, "retry: %d\nevent: put\ndata: {\"path\":\"/\",\"data\":{\"environments\":{},\"filters\":{}}}\n\n", hintMillis) + w.(http.Flusher).Flush() + })) + defer server.Close() + + mockLog := ldlogtest.NewMockLog() + defer mockLog.DumpIfTestFailed(t) + mockLog.Loggers.SetMinLevel(ldlog.Debug) + + httpConfig, err := httpconfig.NewHTTPConfig(config.ProxyConfig{}, config.HTTPConfig{}, nil, "", mockLog.Loggers) + require.NoError(t, err) + serverURL, err := url.Parse(server.URL) + require.NoError(t, err) + + sm := NewStreamManager( + testConfigKey, serverURL, newTestMessageHandler(), httpConfig, + time.Millisecond, rpacProtocolVersion, mockLog.Loggers, noopTestCache{}, + time.Second, true, // ignoreConnectionErrors so Relay stays up and keeps retrying + ) + defer sm.Close() + sm.extendedRetryDelay = 10 * time.Minute // the extended base delay we are supposed to get + + sm.Start() + + // Wait until the key has been rejected several times. + deadline := time.Now().Add(3 * time.Second) + for atomic.LoadInt64(&requestCount) < 5 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + require.GreaterOrEqual(t, atomic.LoadInt64(&requestCount), int64(5), + "the rejecting service was contacted fewer than 5 times") + + mockLog.AssertMessageMatch(t, true, ldlog.Info, "engaging extended backoff") + delays := reconnectDelays(mockLog) + t.Logf("delays eventsource used after the extended profile was engaged: %v", delays) + for _, d := range delays { + assert.Less(t, d, 5*time.Second, + "every delay stayed on the 10ms hint, not the 10 min extended base") + } +} diff --git a/internal/autoconfig/stream_manager_retry_profile_test.go b/internal/autoconfig/stream_manager_retry_profile_test.go new file mode 100644 index 000000000..31ee14b97 --- /dev/null +++ b/internal/autoconfig/stream_manager_retry_profile_test.go @@ -0,0 +1,65 @@ +package autoconfig + +import ( + "net/http" + "net/http/httptest" + "net/url" + "testing" + "time" + + "github.com/launchdarkly/go-sdk-common/v3/ldlog" + "github.com/launchdarkly/go-sdk-common/v3/ldlogtest" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/launchdarkly/ld-relay/v8/config" + "github.com/launchdarkly/ld-relay/v8/internal/httpconfig" +) + +// TestProofExtendedProfilePersistsAcrossTheFirstConnectionLoop is an attempted disproof of +// "the extended profile actually slows retries": it checks whether the activation survives +// repeated passes through eventsource's CanRetryFirstConnection(-1) loop, and whether the +// per-profile counter really doubles from the extended base. It does, so the vector fails. +func TestProofExtendedProfilePersistsAcrossTheFirstConnectionLoop(t *testing.T) { + const base = 20 * time.Millisecond + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(401) + })) + defer server.Close() + + mockLog := ldlogtest.NewMockLog() + defer mockLog.DumpIfTestFailed(t) + mockLog.Loggers.SetMinLevel(ldlog.Debug) + + httpConfig, err := httpconfig.NewHTTPConfig(config.ProxyConfig{}, config.HTTPConfig{}, nil, "", mockLog.Loggers) + require.NoError(t, err) + serverURL, err := url.Parse(server.URL) + require.NoError(t, err) + + sm := NewStreamManager( + testConfigKey, serverURL, newTestMessageHandler(), httpConfig, + time.Millisecond, rpacProtocolVersion, mockLog.Loggers, noopTestCache{}, + time.Second, true, + ) + defer sm.Close() + sm.extendedRetryDelay = base + + sm.Start() + + var delays []time.Duration + require.Eventually(t, func() bool { + delays = reconnectDelays(mockLog) + return len(delays) >= 4 + }, 5*time.Second, 10*time.Millisecond, "expected at least 4 logged delays") + + t.Logf("first four extended delays: %v", delays[:4]) + // Jitter removes up to half, so attempt n's delay is in [base*2^(n-1)/2, base*2^(n-1)]. + for i := 0; i < 4; i++ { + want := time.Duration(1<= 1 }, 2*time.Second, 10*time.Millisecond, + "expected the first attempt") + + sm.Close() + atClose := requests() + + // Several extended delays' worth of wall clock, so a surviving wait would have fired. + time.Sleep(700 * time.Millisecond) + assert.Equal(t, atClose, requests(), "Relay contacted the service after Close() returned") +} + +// hasAnyMessage reports whether any line at the given level contains substr. +func hasAnyMessage(mockLog *ldlogtest.MockLog, level ldlog.LogLevel, substr string) bool { + for _, line := range mockLog.GetOutput(level) { + if strings.Contains(line, substr) { + return true + } + } + return false +} diff --git a/internal/autoconfig/stream_manager_stop_policy_test.go b/internal/autoconfig/stream_manager_stop_policy_test.go new file mode 100644 index 000000000..d51393543 --- /dev/null +++ b/internal/autoconfig/stream_manager_stop_policy_test.go @@ -0,0 +1,98 @@ +package autoconfig + +import ( + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "testing" + "time" + + "github.com/launchdarkly/go-sdk-common/v3/ldlog" + "github.com/launchdarkly/go-sdk-common/v3/ldlogtest" + helpers "github.com/launchdarkly/go-test-helpers/v3" + "github.com/launchdarkly/go-test-helpers/v3/httphelpers" + "github.com/stretchr/testify/require" + + "github.com/launchdarkly/ld-relay/v8/config" + "github.com/launchdarkly/ld-relay/v8/internal/httpconfig" + "github.com/launchdarkly/ld-relay/v8/internal/retry" +) + +// Only a rejected credential can make Relay stop. Other statuses in the same retry class back +// off on the longer delays but must leave Relay running, because a misconfigured stream URI or +// a load balancer answering 404 mid-deploy is not a reason to take a whole fleet down. +// +// This started life as a proof that the opposite happened: an earlier revision routed every +// Unexpected classification to the give-up path, so any of these statuses exited a Relay with +// no cached configuration. +func TestUnexpectedStatusOtherThanRejectionDoesNotStopRelay(t *testing.T) { + for _, status := range []int{404, 405, 410, 418, 451} { + t.Run(fmt.Sprintf("status %d", status), func(t *testing.T) { + // These are all in the slow-retry class, which is the point: the class must not by + // itself decide whether Relay stops. + require.Equal(t, retry.Unexpected, retry.ClassifyHTTPStatus(status)) + + handler, requestsCh := httphelpers.RecordingHandler(httphelpers.HandlerWithStatus(status)) + _, stream := httphelpers.SSEHandler(nil) + defer stream.Close() + + streamManagerTestWithStreamHandler(t, handler, stream, noopTestCache{}, func(p streamManagerTestParams) { + p.streamManager.extendedRetryDelay = time.Millisecond + + readyCh := p.streamManager.Start() + + // It keeps trying rather than reporting a failure. + helpers.RequireValue(t, requestsCh, time.Second, "expected the first attempt") + helpers.RequireValue(t, requestsCh, time.Second, "expected a retry") + + if !helpers.AssertNoMoreValues(t, readyCh, 300*time.Millisecond, + "Relay reported a fatal error for a status that is not a rejected credential") { + t.FailNow() + } + p.mockLog.AssertMessageMatch(t, false, ldlog.Error, "no cached configuration is available") + p.mockLog.AssertMessageMatch(t, false, ldlog.Error, "Invalid auto-configuration key") + }) + }) + } +} + +// A certificate failure is in the slow-retry class too, and likewise must not stop Relay. A +// machine with a skewed clock, or a TLS-terminating proxy mid-cert-rollover, produces this and +// resolves without Relay's involvement. +func TestCertificateFailureDoesNotStopRelay(t *testing.T) { + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(200) + })) + defer server.Close() + + mockLog := ldlogtest.NewMockLog() + defer mockLog.DumpIfTestFailed(t) + mockLog.Loggers.SetMinLevel(ldlog.Debug) + + httpConfig, err := httpconfig.NewHTTPConfig(config.ProxyConfig{}, config.HTTPConfig{}, nil, "", mockLog.Loggers) + require.NoError(t, err) + serverURL, err := url.Parse(server.URL) + require.NoError(t, err) + + sm := NewStreamManager( + testConfigKey, serverURL, newTestMessageHandler(), httpConfig, + time.Millisecond, rpacProtocolVersion, mockLog.Loggers, noopTestCache{}, + time.Second, false, + ) + defer sm.Close() + sm.extendedRetryDelay = time.Millisecond + + readyCh := sm.Start() + if !helpers.AssertNoMoreValues(t, readyCh, time.Second, + "Relay reported a fatal error for a certificate failure") { + t.FailNow() + } + mockLog.AssertMessageMatch(t, false, ldlog.Error, "no cached configuration is available") + + // A certificate failure is a transport failure, and no transport failure is unexpected, so + // it must stay on the short delays. Waiting minutes would not help: the failure resolves + // the moment an operator fixes the certificate. + mockLog.AssertMessageMatch(t, false, ldlog.Info, "engaging extended backoff") + mockLog.AssertMessageMatch(t, true, ldlog.Warn, "Unexpected error on auto-configuration stream") +} diff --git a/relay/relay.go b/relay/relay.go index 9a6d25b66..5fdf7a9ed 100644 --- a/relay/relay.go +++ b/relay/relay.go @@ -229,11 +229,12 @@ func newRelayInternal(c config.Config, options relayInternalOptions) (*Relay, er // affect this; a failure of auto-config is more serious than any // environment-specific failure. // - // A rejected auto-configuration key no longer reaches this point on its own. The - // StreamManager keeps retrying, and reports a failure here only after a grace - // period during which no configuration arrived from either the stream or the - // persistent cache. With a cached configuration Relay stays running and serves - // from it. The StreamManager has already logged the reason by this point. + // A rejected auto-configuration key reaches this point only when Relay has + // nothing to serve: the stream never delivered a configuration and the + // persistent cache is reachable and empty. With a cached configuration Relay + // stays running, serves from it, and keeps retrying the key. Any failure other + // than a rejected key also leaves Relay running. The StreamManager has already + // logged the reason by this point. os.Exit(1) } }() From 11ed0c6ce5afe56833f9136cf0bb68c119c934ce Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Tue, 15 Sep 2026 12:29:49 -0400 Subject: [PATCH 4/4] fix: Report a rejected key as interrupted rather than off The stream keeps retrying a rejected credential, so it is not finished. Off is left for the two cases that do finish it: Close, and giving up because there is no configuration to serve. This is the one status state that differs from the reporting this is built on, where a rejected key is terminal because the stream stops. --- internal/autoconfig/stream_manager.go | 24 ++++++++++++++ .../stream_manager_stop_policy_test.go | 31 +++++++++++++++++++ 2 files changed, 55 insertions(+) diff --git a/internal/autoconfig/stream_manager.go b/internal/autoconfig/stream_manager.go index dfe1f282c..dedb3d59c 100644 --- a/internal/autoconfig/stream_manager.go +++ b/internal/autoconfig/stream_manager.go @@ -364,6 +364,7 @@ func (s *StreamManager) subscribe(readyCh chan<- error) { // presenting a credential the service has already rejected, for as long as the // extended delays run. s.streamCancel() + s.updateStatus(interfaces.DataSourceStateOff, interfaces.DataSourceErrorInfo{}) s.abandonStreamGoroutine(streamCh) } @@ -661,6 +662,10 @@ func (s *StreamManager) newStreamErrorHandler( class, keyRejected := s.classifyAndLogStreamError(err) + // Interrupted rather than Off even for a rejected credential: the stream keeps + // retrying, so it is not finished. Off is left for Close and for giving up. + s.updateStatus(interfaces.DataSourceStateInterrupted, streamErrorInfo(err)) + result := es.StreamErrorHandlerResult{CloseNow: false} if class == retry.Unexpected { if !loggedExtended { @@ -841,6 +846,25 @@ func obfuscateEventData(data string) string { return data } +// streamErrorInfo describes a stream failure in the shape the SDK data source status uses, so +// the status resource reports both the same way. An HTTP failure carries its status code; a +// transport failure has none to carry. +func streamErrorInfo(err error) interfaces.DataSourceErrorInfo { + var se es.SubscriptionError + if errors.As(err, &se) { + return interfaces.DataSourceErrorInfo{ + Kind: interfaces.DataSourceErrorKindErrorResponse, + StatusCode: se.Code, + Time: time.Now(), + } + } + return interfaces.DataSourceErrorInfo{ + Kind: interfaces.DataSourceErrorKindNetworkError, + Message: err.Error(), + Time: time.Now(), + } +} + // StreamStatus is the state of the auto-configuration stream connection. It uses the same types // as the SDK data source status, so that the two report the same states and error kinds. type StreamStatus struct { diff --git a/internal/autoconfig/stream_manager_stop_policy_test.go b/internal/autoconfig/stream_manager_stop_policy_test.go index d51393543..054f40202 100644 --- a/internal/autoconfig/stream_manager_stop_policy_test.go +++ b/internal/autoconfig/stream_manager_stop_policy_test.go @@ -10,8 +10,10 @@ import ( "github.com/launchdarkly/go-sdk-common/v3/ldlog" "github.com/launchdarkly/go-sdk-common/v3/ldlogtest" + "github.com/launchdarkly/go-server-sdk/v7/interfaces" helpers "github.com/launchdarkly/go-test-helpers/v3" "github.com/launchdarkly/go-test-helpers/v3/httphelpers" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/launchdarkly/ld-relay/v8/config" @@ -96,3 +98,32 @@ func TestCertificateFailureDoesNotStopRelay(t *testing.T) { mockLog.AssertMessageMatch(t, false, ldlog.Info, "engaging extended backoff") mockLog.AssertMessageMatch(t, true, ldlog.Warn, "Unexpected error on auto-configuration stream") } + +// The status delta this change introduces: a rejected credential is no longer terminal, because +// the stream keeps retrying. On the branch this is stacked on, the same rejection reports OFF. +// +// OFF now means only that the stream is finished: Close was called, or Relay gave up because it +// had nothing to serve. +func TestRejectedKeyIsNotTerminalInTheStatus(t *testing.T) { + handler := httphelpers.HandlerWithStatus(401) + _, stream := httphelpers.SSEHandler(nil) + defer stream.Close() + + streamManagerTestWithStreamHandler(t, handler, stream, noopTestCache{}, func(p streamManagerTestParams) { + p.streamManager.extendedRetryDelay = time.Millisecond + // Without this Relay would give up, which is terminal for a different reason. + p.streamManager.ignoreConnectionErrors = true + p.streamManager.Start() + + require.Eventually(t, func() bool { + return p.streamManager.Status().LastError.StatusCode == 401 + }, 2*time.Second, 10*time.Millisecond, "expected the rejection to be recorded") + + st := p.streamManager.Status() + assert.NotEqual(t, interfaces.DataSourceStateOff, st.State, + "the stream is still retrying, so it is not finished") + // It never connected, so an interruption keeps the initializing state. + assert.Equal(t, interfaces.DataSourceStateInitializing, st.State) + assert.Equal(t, interfaces.DataSourceErrorKindErrorResponse, st.LastError.Kind) + }) +}