feat: conform FDv1 streaming and polling to RETRY spec

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/google/uuid v1.1.1
 	github.com/gregjones/httpcache v0.0.0-20171119193500-2bcd89a1743f
 	github.com/launchdarkly/ccache v1.1.0
-	github.com/launchdarkly/eventsource v1.10.0
+	github.com/launchdarkly/eventsource v1.13.0
 	github.com/launchdarkly/go-jsonstream/v3 v3.1.1
 	github.com/launchdarkly/go-ntlm-proxy-auth v1.0.3
 	github.com/launchdarkly/go-sdk-common/v3 v3.5.0

go.sum

@@ -17,8 +17,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/launchdarkly/ccache v1.1.0 h1:voD1M+ZJXR3MREOKtBwgTF9hYHl1jg+vFKS/+VAkR2k=
 github.com/launchdarkly/ccache v1.1.0/go.mod h1:TlxzrlnzvYeXiLHmesMuvoZetu4Z97cV1SsdqqBJi1Q=
-github.com/launchdarkly/eventsource v1.10.0 h1:H9Tp6AfGu/G2qzBJC26iperrvwhzdbiA/gx7qE2nDFI=
-github.com/launchdarkly/eventsource v1.10.0/go.mod h1:J3oa50bPvJesZqNAJtb5btSIo5N6roDWhiAS3IpsKck=
+github.com/launchdarkly/eventsource v1.13.0 h1:SjC1LgNSR+ip8BgZcphGa/ar6N0I+pFTYQ/1ZzDnP98=
+github.com/launchdarkly/eventsource v1.13.0/go.mod h1:dU+rZxkPOlGPsyJPpiDqiepAcFwIITDUClY9+A6RrMw=
 github.com/launchdarkly/go-jsonstream/v3 v3.1.1 h1:ugupp2eNtwVbr69KCdeUrm1vUf1/3ju4Wdliaob95uY=
 github.com/launchdarkly/go-jsonstream/v3 v3.1.1/go.mod h1:ZBjhKq8mhArCtqotGRGnteY6eXpNm1GaOdUSZHh+ZjM=
 github.com/launchdarkly/go-ntlm-proxy-auth v1.0.3 h1:i3V0N+R0Fd2nXfGEVKCBIZ8kyttZ+SRKvBG8cdcphO4=

internal/datasource/helpers.go

@@ -1,6 +1,9 @@
 package datasource
 
 import (
+	"crypto/tls"
+	"crypto/x509"
+	"errors"
 	"fmt"
 	"net/http"
 
@@ -20,53 +23,100 @@ func (e httpStatusError) Error() string {
 	return e.Message
 }
 
-// Tests whether an HTTP error status represents a condition that might resolve on its own if we retry,
-// or at least should not make us permanently stop sending requests.
-func isHTTPErrorRecoverable(statusCode int) bool {
+// FailureClass categorizes a data source failure per RETRY §1.5--§1.7. Under the
+// RETRY spec no failure is permanently terminal: every failure is either "normal"
+// (regular backoff and retry) or "unexpected" (extended backoff via a longer
+// retry profile or wait interval, still retrying indefinitely).
+type FailureClass int
+
+const (
+	// FailureClassNormal indicates a failure that a caller should treat as an
+	// ordinary transient error.
+	FailureClassNormal FailureClass = iota
+	// FailureClassUnexpected indicates a failure that the caller should treat as
+	// signalling a durable, non-transient upstream problem.
+	FailureClassUnexpected
+)
+
+// classifyHTTPFailure returns the failure classification for an HTTP status code
+// received during a data source request, per RETRY §1.6. Called only when the
+// status indicates failure (non-2xx).
+func classifyHTTPFailure(statusCode int) FailureClass {
 	if statusCode >= 400 && statusCode < 500 {
 		switch statusCode {
-		case 400: // bad request
-			return true
-		case 408: // request timeout
-			return true
-		case 429: // too many requests
-			return true
+		case 400, 408, 429:
+			return FailureClassNormal
 		default:
-			return false // all other 4xx errors are unrecoverable
+			return FailureClassUnexpected
 		}
 	}
-	return true
+	return FailureClassNormal
+}
+
+// classifyTransportFailure returns the failure classification for a transport-layer
+// error (i.e., not an HTTP response, but a lower-level network or TLS failure)
+// per RETRY §1.7. TLS/certificate validation failures are treated as unexpected;
+// everything else is treated as normal.
+func classifyTransportFailure(err error) FailureClass {
+	if err == nil {
+		return FailureClassNormal
+	}
+	var certErr *tls.CertificateVerificationError
+	if errors.As(err, &certErr) {
+		return FailureClassUnexpected
+	}
+	var x509UnknownAuthorityErr x509.UnknownAuthorityError
+	if errors.As(err, &x509UnknownAuthorityErr) {
+		return FailureClassUnexpected
+	}
+	var x509HostnameErr x509.HostnameError
+	if errors.As(err, &x509HostnameErr) {
+		return FailureClassUnexpected
+	}
+	var x509InvalidErr x509.CertificateInvalidError
+	if errors.As(err, &x509InvalidErr) {
+		return FailureClassUnexpected
+	}
+	return FailureClassNormal
 }
 
 func httpErrorDescription(statusCode int) string {
 	message := ""
 	if statusCode == 401 || statusCode == 403 {
-		message = " (invalid SDK key)"
+		message = " (authentication failed)"
 	}
 	return fmt.Sprintf("HTTP error %d%s", statusCode, message)
 }
 
-// Logs an HTTP error or network error at the appropriate level and determines whether it is recoverable
-// (as defined by isHTTPErrorRecoverable).
-func checkIfErrorIsRecoverableAndLog(
+// classifyAndLogHTTPFailure classifies an HTTP failure per RETRY §1.6, logs it,
+// and returns the classification for the caller to act on.
+func classifyAndLogHTTPFailure(
 	loggers ldlog.Loggers,
 	errorDesc, errorContext string,
 	statusCode int,
-	recoverableMessage string,
-) bool {
-	if statusCode > 0 && !isHTTPErrorRecoverable(statusCode) {
-		loggers.Errorf("Error %s (giving up permanently): %s", errorContext, errorDesc)
-		return false
-	}
-	loggers.Warnf("Error %s (%s): %s", errorContext, recoverableMessage, errorDesc)
-	return true
+	willRetryMessage string,
+) FailureClass {
+	loggers.Warnf("Error %s (%s): %s", errorContext, willRetryMessage, errorDesc)
+	return classifyHTTPFailure(statusCode)
+}
+
+// classifyAndLogTransportFailure classifies a transport-layer failure per RETRY
+// §1.7, logs it, and returns the classification.
+func classifyAndLogTransportFailure(
+	loggers ldlog.Loggers,
+	err error,
+	errorContext, willRetryMessage string,
+) FailureClass {
+	loggers.Warnf("Error %s (%s): %s", errorContext, willRetryMessage, err.Error())
+	return classifyTransportFailure(err)
 }
 
 func checkForHTTPError(statusCode int, url string) error {
 	if statusCode == http.StatusUnauthorized {
 		return httpStatusError{
-			Message: fmt.Sprintf("Invalid SDK key when accessing URL: %s. Verify that your SDK key is correct.", url),
-			Code:    statusCode}
+			Message: fmt.Sprintf("Authentication failed for URL: %s. If this persists, verify that your SDK key is correct.",
+				url),
+			Code: statusCode}
 	}
 
 	if statusCode == http.StatusNotFound {

internal/datasource/helpers_test.go

@@ -1,6 +1,10 @@
 package datasource
 
 import (
+	"crypto/tls"
+	"crypto/x509"
+	"errors"
+	"fmt"
 	"strconv"
 	"testing"
 
@@ -12,19 +16,47 @@ func TestHTTPStatusError(t *testing.T) {
 	assert.Equal(t, "message", error.Error())
 }
 
-func TestIsHTTPErrorRecoverable(t *testing.T) {
+// classifyHTTPFailure per RETRY §1.6: 400, 408, 429 are normal (transient
+// server-side conditions); all other 4xx are unexpected (durable client-side
+// misconfiguration); 5xx and everything else are normal.
+func TestClassifyHTTPFailure(t *testing.T) {
 	for i := 400; i < 500; i++ {
-		assert.Equal(t, i == 400 || i == 408 || i == 429, isHTTPErrorRecoverable(i), strconv.Itoa(i))
+		expected := FailureClassNormal
+		if !(i == 400 || i == 408 || i == 429) {
+			expected = FailureClassUnexpected
+		}
+		assert.Equal(t, expected, classifyHTTPFailure(i), strconv.Itoa(i))
 	}
 	for i := 500; i < 600; i++ {
-		assert.True(t, isHTTPErrorRecoverable(i))
+		assert.Equal(t, FailureClassNormal, classifyHTTPFailure(i), strconv.Itoa(i))
 	}
 }
 
+// classifyTransportFailure per RETRY §1.7: TLS/certificate validation failures
+// are unexpected; other transport-layer errors are normal.
+func TestClassifyTransportFailure(t *testing.T) {
+	assert.Equal(t, FailureClassNormal, classifyTransportFailure(nil))
+	assert.Equal(t, FailureClassNormal, classifyTransportFailure(errors.New("boom")))
+
+	// TLS certificate errors.
+	assert.Equal(t, FailureClassUnexpected,
+		classifyTransportFailure(&tls.CertificateVerificationError{}))
+	assert.Equal(t, FailureClassUnexpected,
+		classifyTransportFailure(x509.UnknownAuthorityError{}))
+	assert.Equal(t, FailureClassUnexpected,
+		classifyTransportFailure(x509.HostnameError{Host: "example.invalid"}))
+	assert.Equal(t, FailureClassUnexpected,
+		classifyTransportFailure(x509.CertificateInvalidError{Reason: x509.Expired}))
+
+	// Wrapped errors are still classified correctly.
+	wrapped := fmt.Errorf("some wrapper: %w", x509.UnknownAuthorityError{})
+	assert.Equal(t, FailureClassUnexpected, classifyTransportFailure(wrapped))
+}
+
 func TestHTTPErrorDescription(t *testing.T) {
 	assert.Equal(t, "HTTP error 400", httpErrorDescription(400))
-	assert.Equal(t, "HTTP error 401 (invalid SDK key)", httpErrorDescription(401))
-	assert.Equal(t, "HTTP error 403 (invalid SDK key)", httpErrorDescription(403))
+	assert.Equal(t, "HTTP error 401 (authentication failed)", httpErrorDescription(401))
+	assert.Equal(t, "HTTP error 403 (authentication failed)", httpErrorDescription(403))
 	assert.Equal(t, "HTTP error 500", httpErrorDescription(500))
 }
 

internal/datasource/polling_data_source.go

@@ -21,9 +21,10 @@ const (
 // PollingConfig describes the configuration for a polling data source. It is exported so that
 // it can be used in the PollingDataSourceBuilder.
 type PollingConfig struct {
-	BaseURI      string
-	PollInterval time.Duration
-	FilterKey    string
+	BaseURI                     string
+	PollInterval                time.Duration
+	FilterKey                   string
+	ExtendedInitialPollInterval time.Duration
 }
 
 // Requester allows PollingProcessor to delegate fetching data to another component.
@@ -43,6 +44,7 @@ type PollingProcessor struct {
 	dataSourceUpdates  subsystems.DataSourceUpdateSink
 	requester          Requester
 	pollInterval       time.Duration
+	strategy           *pollingStrategy
 	loggers            ldlog.Loggers
 	setInitializedOnce sync.Once
 	isInitialized      internal.AtomicBoolean
@@ -57,19 +59,24 @@ func NewPollingProcessor(
 	cfg PollingConfig,
 ) *PollingProcessor {
 	httpRequester := NewPollingRequester(context, context.GetHTTP().CreateHTTPClient(), cfg.BaseURI, cfg.FilterKey)
-	return newPollingProcessor(context, dataSourceUpdates, httpRequester, cfg.PollInterval)
+	return newPollingProcessor(
+		context, dataSourceUpdates, httpRequester,
+		cfg.PollInterval, cfg.ExtendedInitialPollInterval,
+	)
 }
 
 func newPollingProcessor(
 	context subsystems.ClientContext,
 	dataSourceUpdates subsystems.DataSourceUpdateSink,
 	requester Requester,
 	pollInterval time.Duration,
+	extendedInitialPollInterval time.Duration,
 ) *PollingProcessor {
 	pp := &PollingProcessor{
 		dataSourceUpdates: dataSourceUpdates,
 		requester:         requester,
 		pollInterval:      pollInterval,
+		strategy:          newPollingStrategy(pollInterval, extendedInitialPollInterval),
 		loggers:           context.GetLogging().Loggers,
 		quit:              make(chan struct{}),
 	}
@@ -80,10 +87,13 @@ func newPollingProcessor(
 func (pp *PollingProcessor) Start(closeWhenReady chan<- struct{}) {
 	pp.loggers.Infof("Starting LaunchDarkly polling with interval: %+v", pp.pollInterval)
 
-	ticker := newTickerWithInitialTick(pp.pollInterval)
+	// Fires immediately for the first poll; Reset after each iteration to schedule
+	// the next. Under RETRY (SDK-2775), the interval between polls is dynamic per
+	// the pollingStrategy state machine, so we can't use a fixed-period Ticker.
+	timer := time.NewTimer(0)
 
 	go func() {
-		defer ticker.Stop()
+		defer timer.Stop()
 
 		var readyOnce sync.Once
 		notifyReady := func() {
@@ -98,28 +108,23 @@ func (pp *PollingProcessor) Start(closeWhenReady chan<- struct{}) {
 			select {
 			case <-pp.quit:
 				return
-			case <-ticker.C:
+			case <-timer.C:
 				if err := pp.poll(); err != nil {
+					var class FailureClass
 					if hse, ok := err.(httpStatusError); ok {
 						errorInfo := interfaces.DataSourceErrorInfo{
 							Kind:       interfaces.DataSourceErrorKindErrorResponse,
 							StatusCode: hse.Code,
 							Time:       time.Now(),
 						}
-						recoverable := checkIfErrorIsRecoverableAndLog(
+						class = classifyAndLogHTTPFailure(
 							pp.loggers,
 							httpErrorDescription(hse.Code),
 							pollingErrorContext,
 							hse.Code,
 							pollingWillRetryMessage,
 						)
-						if recoverable {
-							pp.dataSourceUpdates.UpdateStatus(interfaces.DataSourceStateInterrupted, errorInfo)
-						} else {
-							pp.dataSourceUpdates.UpdateStatus(interfaces.DataSourceStateOff, errorInfo)
-							notifyReady()
-							return
-						}
+						pp.dataSourceUpdates.UpdateStatus(interfaces.DataSourceStateInterrupted, errorInfo)
 					} else {
 						errorInfo := interfaces.DataSourceErrorInfo{
 							Kind:    interfaces.DataSourceErrorKindNetworkError,
@@ -129,17 +134,24 @@ func (pp *PollingProcessor) Start(closeWhenReady chan<- struct{}) {
 						if _, ok := err.(malformedJSONError); ok {
 							errorInfo.Kind = interfaces.DataSourceErrorKindInvalidData
 						}
-						checkIfErrorIsRecoverableAndLog(pp.loggers, err.Error(), pollingErrorContext, 0, pollingWillRetryMessage)
+						class = classifyAndLogTransportFailure(
+							pp.loggers, err, pollingErrorContext, pollingWillRetryMessage,
+						)
 						pp.dataSourceUpdates.UpdateStatus(interfaces.DataSourceStateInterrupted, errorInfo)
 					}
-					continue
+					if pp.strategy.OnFailure(class) {
+						pp.loggers.Info("Classified failure as UNEXPECTED; engaging extended backoff.")
+					}
+				} else {
+					pp.dataSourceUpdates.UpdateStatus(interfaces.DataSourceStateValid, interfaces.DataSourceErrorInfo{})
+					pp.setInitializedOnce.Do(func() {
+						pp.isInitialized.Set(true)
+						pp.loggers.Info("First polling request successful")
+						notifyReady()
+					})
+					pp.strategy.OnSuccess()
 				}
-				pp.dataSourceUpdates.UpdateStatus(interfaces.DataSourceStateValid, interfaces.DataSourceErrorInfo{})
-				pp.setInitializedOnce.Do(func() {
-					pp.isInitialized.Set(true)
-					pp.loggers.Info("First polling request successful")
-					notifyReady()
-				})
+				timer.Reset(pp.strategy.NextWait())
 			}
 		}
 	}()
@@ -190,24 +202,3 @@ func (pp *PollingProcessor) GetPollInterval() time.Duration {
 func (pp *PollingProcessor) GetFilterKey() string {
 	return pp.requester.FilterKey()
 }
-
-type tickerWithInitialTick struct {
-	*time.Ticker
-	C <-chan time.Time
-}
-
-func newTickerWithInitialTick(interval time.Duration) *tickerWithInitialTick {
-	c := make(chan time.Time)
-	ticker := time.NewTicker(interval)
-	t := &tickerWithInitialTick{
-		C:      c,
-		Ticker: ticker,
-	}
-	go func() {
-		c <- time.Now() // Ensure we do an initial poll immediately
-		for tt := range ticker.C {
-			c <- tt
-		}
-	}()
-	return t
-}