diff --git a/docs/proxy-mode.md b/docs/proxy-mode.md index d3f3316e..a3a62ecd 100644 --- a/docs/proxy-mode.md +++ b/docs/proxy-mode.md @@ -52,6 +52,22 @@ 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. Restart the Relay Proxy if you need it to reconnect immediately. + +This is the same thing the Relay Proxy already did for every other failure to reach LaunchDarkly, such as a `404`, a `5xx`, a network error, or a certificate problem. A rejected key is no longer treated differently. + +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 answers every request with a `503` error until the key becomes valid. + +Read the Relay Proxy logs, or the [status resource](./endpoints.md), to tell a rejected key apart from an unreachable LaunchDarkly service. A rejected key logs `Invalid auto-configuration key; will keep retrying in case it becomes valid`, and the status resource reports the auto-configuration stream as `INTERRUPTED` with the status code. + +Versions of the Relay Proxy before 8.22.0 shut the process down immediately when LaunchDarkly rejected the auto-configuration key, even when a persistent store held usable configuration data. If you relied on the process exiting to detect a bad key, use the status resource instead. + ### 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 97101b06..6d6ba0e0 100644 --- a/internal/autoconfig/errors_and_messages.go +++ b/internal/autoconfig/errors_and_messages.go @@ -5,7 +5,8 @@ const ( 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." 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 b92ca99e..76a05103 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,19 @@ 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 a failure looks unlikely to correct + // itself soon. + extendedRetryDelay time.Duration + loggers ldlog.Loggers + halt chan 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. @@ -118,16 +136,22 @@ 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, 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, + loggers: loggers, + halt: make(chan struct{}), status: StreamStatus{ State: interfaces.DataSourceStateInitializing, StateSince: time.Now(), @@ -191,6 +215,11 @@ 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. Without + // this, Close would block on s.done for the remainder of a delay that now reaches an + // hour. + s.streamCancel() s.updateStatus(interfaces.DataSourceStateOff, interfaces.DataSourceErrorInfo{}) }) if s.done != nil { @@ -218,44 +247,22 @@ 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), + ) + errorHandler := s.newStreamErrorHandler(extendedProfile) rpacEndpoint, err := url.JoinPath(s.uri.String(), autoConfigStreamPath) if err != nil { @@ -264,7 +271,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) @@ -279,9 +286,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), @@ -549,6 +555,87 @@ func (s *StreamManager) dispatchFilterAction(id config.FilterID, rep envfactory. } } +// newStreamErrorHandler builds the SSE error handler for one subscribe cycle. +// +// Nothing stops the stream. A rejected key can become valid again without Relay knowing, and +// every other failure could clear at any time, so the stream keeps retrying in all cases. A +// failure that is unlikely to correct itself soon moves to the longer delays, which bounds the +// load a fleet puts on a service that is rejecting every request. +func (s *StreamManager) newStreamErrorHandler(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: + } + + // Interrupted rather than Off even for a rejected key: the stream keeps retrying, so + // it is not finished. Off is left for Close. + s.updateStatus(interfaces.DataSourceStateInterrupted, streamErrorInfo(err)) + + result := es.StreamErrorHandlerResult{CloseNow: false} + if s.classifyAndLogStreamError(err) == retry.Unexpected { + if !loggedExtended { + s.loggers.Info(logMsgExtendedBackoff) + loggedExtended = true + } + result.ActivateProfile = extendedProfile + } + return result + } +} + +// classifyAndLogStreamError sorts a stream failure into a retry class and logs it. The class +// decides how long to wait before the next attempt; it never decides whether to keep trying. +// +// 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) retry.FailureClass { + var se es.SubscriptionError + if !errors.As(err, &se) { + // No transport-level failure is unexpected, so these keep the short delays. + s.loggers.Warnf(logMsgStreamOtherError, err) + return retry.Normal + } + + class := retry.ClassifyHTTPStatus(se.Code) + switch { + case se.Code == http.StatusUnauthorized || se.Code == http.StatusForbidden: + s.loggers.Error(logMsgBadKeyWillRetry) + case class == retry.Unexpected: + s.loggers.Errorf(logMsgStreamHTTPError, se.Code) + default: + s.loggers.Warnf(logMsgStreamHTTPError, se.Code) + } + return class +} + +// 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(), + } +} + func (s *StreamManager) applyCachedContent(content *PutContent) { s.handlePut(PutContent{ Environments: content.Environments, @@ -720,8 +807,8 @@ func (s *StreamManager) setStatus(state interfaces.DataSourceState, errorInfo in } if s.status.State == interfaces.DataSourceStateOff { - // OFF is terminal: Close was called, or the key was rejected and the stream will not be - // retried. An event already in flight must not report it as working. + // OFF is terminal: Close was called, so the stream is finished. An event already in + // flight must not report it as working. return } diff --git a/internal/autoconfig/stream_manager_errors_test.go b/internal/autoconfig/stream_manager_errors_test.go index 19fdcd1d..dd642495 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,84 @@ 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 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 TestServesTheCachedConfigurationWhileTheKeyIsRejected(t *testing.T) { + // A rejected key leaves Relay running, so a cached configuration is worth having: Relay + // serves those environments while it keeps trying the key. Before this change the process + // exited and threw the cache away. + 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() + + if !helpers.AssertNoMoreValues(t, readyCh, time.Second, "Relay reported a failure") { + t.FailNow() + } + p.mockLog.AssertMessageMatch(t, true, ldlog.Info, "loaded from persistent cache") + }) +} + +func TestKeepsRunningWithNoConfigurationAtAll(t *testing.T) { + // The case that used to exit the process. With a rejected key and nothing cached, Relay has + // nothing to serve, and it still keeps running and retrying rather than reporting a failure. + // It answers 503 until the key becomes valid, which is what it already did for every other + // failure to reach LaunchDarkly. + 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() + + if !helpers.AssertNoMoreValues(t, readyCh, 500*time.Millisecond, + "Relay reported a failure on a rejected key") { + t.FailNow() + } + p.mockLog.AssertMessageMatch(t, true, ldlog.Error, "will keep retrying") + }) +} 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 00000000..108ee4e2 --- /dev/null +++ b/internal/autoconfig/stream_manager_known_limits_test.go @@ -0,0 +1,105 @@ +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" + "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" +) + +// 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{}, + ) + 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 00000000..df964746 --- /dev/null +++ b/internal/autoconfig/stream_manager_retry_profile_test.go @@ -0,0 +1,67 @@ +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" +) + +// TestExtendedDelaysDoubleFromTheExtendedBase checks that engaging the extended profile really +// slows retries down, which is the point of the change and the part no other test covers: +// removing the profile activation leaves every other test in the package green. +// +// The activation has to survive repeated passes through eventsource's +// CanRetryFirstConnection(-1) loop, and the per-profile attempt counter has to double from the +// extended base rather than the normal one. Both hold. +func TestExtendedDelaysDoubleFromTheExtendedBase(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{}, + ) + 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_status_test.go b/internal/autoconfig/stream_manager_status_test.go index 646e62c4..dbf89e66 100644 --- a/internal/autoconfig/stream_manager_status_test.go +++ b/internal/autoconfig/stream_manager_status_test.go @@ -5,6 +5,7 @@ import ( "time" "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" @@ -63,25 +64,36 @@ func TestStatusRecordsTheHTTPStatusCode(t *testing.T) { }) } -// A rejected key is terminal on this line: the stream stops and Relay exits, so the state is OFF -// and nothing can move it afterwards. -func TestStatusIsOffAfterARejectedKey(t *testing.T) { +// A rejected key is not terminal: the stream keeps retrying, so it records the failure without +// reporting the stream as finished, and it reports nothing on the ready channel. Before this +// change the state went to OFF and Relay exited. +func TestStatusIsNotOffAfterARejectedKey(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 readyCh := p.streamManager.Start() - err := <-readyCh - require.Error(t, err) + + 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.Equal(t, interfaces.DataSourceStateOff, st.State) - assert.Equal(t, 401, st.LastError.StatusCode) + 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) + + if !helpers.AssertNoMoreValues(t, readyCh, 300*time.Millisecond, "Relay reported a failure") { + t.FailNow() + } }) } -// Close is terminal too. +// Close is what makes the state terminal now. func TestStatusIsOffAfterClose(t *testing.T) { initialEvent := makeEnvPutEvent(testEnv1) streamManagerTest(t, &initialEvent, func(p streamManagerTestParams) { 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 00000000..b577c137 --- /dev/null +++ b/internal/autoconfig/stream_manager_stop_policy_test.go @@ -0,0 +1,91 @@ +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" +) + +// Every status in the slow-retry class keeps Relay running and keeps it trying. The class +// decides how long to wait, and nothing else: a misconfigured stream URI or a load balancer +// answering 404 mid-deploy must not take a fleet down, and neither must a rejected key. +func TestUnexpectedStatusKeepsRelayRunning(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 picks the + // delay and never decides whether to keep trying. + 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, "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{}, + ) + 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() + } + + // 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") +}