feat: Keep retrying a rejected auto-configuration key

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.

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"

internal/autoconfig/stream_manager.go

@@ -18,6 +18,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 (
@@ -28,6 +29,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 (
@@ -77,10 +86,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.
@@ -102,6 +119,8 @@ func NewStreamManager(
 	protocolVersion int,
 	loggers ldlog.Loggers,
 	cache Cache,
+	initTimeout time.Duration,
+	ignoreConnectionErrors bool,
 ) *StreamManager {
 	loggers.SetPrefix("AutoConfiguration")
 	if protocolVersion > 1 {
@@ -117,8 +136,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{}),
 	}
 
 	// Enforces ordering constraints on the SSE messages that are sent from the server, allowing the MessageHandler
@@ -204,32 +228,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 {
-			if se.Code == 401 || se.Code == 403 {
-				s.loggers.Error(logMsgBadKey)
-				signalReady(errors.New("invalid auto-configuration key"))
-				return es.StreamErrorHandlerResult{CloseNow: true}
-			}
-			s.loggers.Warnf(logMsgStreamHTTPError, se.Code)
-			return es.StreamErrorHandlerResult{CloseNow: false}
-		}
-
-		s.loggers.Warnf(logMsgStreamOtherError, err)
-		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 {
@@ -253,9 +271,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),
@@ -267,18 +284,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 {
@@ -294,14 +368,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
 		}
 	}
@@ -516,6 +583,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,