fix(flagd): do not retry for certain status codes (#756)

Makefile

@@ -17,7 +17,7 @@ test:
 
 # call with TESTCONTAINERS_RYUK_DISABLED="true" to avoid problems with podman on Macs
 e2e:
-	go clean -testcache && go list -f '{{.Dir}}/...' -m | xargs -I{} go test -timeout=2m -tags=e2e {}
+	go clean -testcache && go list -f '{{.Dir}}/...' -m | xargs -I{} go test -timeout=3m -tags=e2e {}
 
 lint:
 	go install -v github.com/golangci/golangci-lint/v2/cmd/golangci-lint@$(GOLANGCI_LINT_VERSION)

providers/flagd/e2e/inprocess_test.go

@@ -3,6 +3,7 @@
 package e2e
 
 import (
+	flagd "github.com/open-feature/go-sdk-contrib/providers/flagd/pkg"
 	"testing"
 
 	"github.com/open-feature/go-sdk-contrib/tests/flagd/testframework"
@@ -17,6 +18,9 @@ func TestInProcessProviderE2E(t *testing.T) {
 	runner := testframework.NewTestbedRunner(testframework.TestbedConfig{
 		ResolverType:  testframework.InProcess,
 		TestbedConfig: "default",
+		ExtraOptions: []flagd.ProviderOption{
+			flagd.WithRetryBackoffMaxMs(5000),
+		},
 	})
 	defer runner.Cleanup()
 

providers/flagd/e2e/rpc_test.go

@@ -26,7 +26,7 @@ func TestRPCProviderE2E(t *testing.T) {
 	}
 
 	// Run tests with RPC-specific tags - exclude unimplemented scenarios
-	tags := "@rpc && ~@unixsocket && ~@targetURI && ~@sync && ~@metadata && ~@grace && ~@customCert && ~@caching"
+	tags := "@rpc && ~@unixsocket && ~@targetURI && ~@sync && ~@metadata && ~@grace && ~@customCert && ~@caching && ~@forbidden"
 
 	if err := runner.RunGherkinTestsWithSubtests(t, featurePaths, tags); err != nil {
 		t.Fatalf("Gherkin tests failed: %v", err)

providers/flagd/pkg/configuration.go

@@ -26,6 +26,9 @@ const (
 	defaultHost                         = "localhost"
 	defaultResolver                     = rpc
 	defaultGracePeriod                  = 5
+	defaultRetryBackoffMs               = 1000
+	defaultRetryBackoffMaxMs            = 120000
+	defaultFatalStatusCodes             = "UNAUTHENTICATED,PERMISSION_DENIED"
 
 	rpc       ResolverType = "rpc"
 	inProcess ResolverType = "in-process"
@@ -45,6 +48,9 @@ const (
 	flagdOfflinePathEnvironmentVariableName           = "FLAGD_OFFLINE_FLAG_SOURCE_PATH"
 	flagdTargetUriEnvironmentVariableName             = "FLAGD_TARGET_URI"
 	flagdGracePeriodVariableName                      = "FLAGD_RETRY_GRACE_PERIOD"
+	flagdRetryBackoffMsVariableName                   = "FLAGD_RETRY_BACKOFF_MS"
+	flagdRetryBackoffMaxMsVariableName                = "FLAGD_RETRY_BACKOFF_MAX_MS"
+	flagdFatalStatusCodesVariableName                 = "FLAGD_FATAL_STATUS_CODES"
 )
 
 type ProviderConfiguration struct {
@@ -66,6 +72,9 @@ type ProviderConfiguration struct {
 	CustomSyncProviderUri            string
 	GrpcDialOptionsOverride          []grpc.DialOption
 	RetryGracePeriod                 int
+	RetryBackoffMs                   int
+	RetryBackoffMaxMs                int
+	FatalStatusCodes                 []string
 
 	log logr.Logger
 }
@@ -80,6 +89,9 @@ func newDefaultConfiguration(log logr.Logger) *ProviderConfiguration {
 		Resolver:                         defaultResolver,
 		Tls:                              defaultTLS,
 		RetryGracePeriod:                 defaultGracePeriod,
+		RetryBackoffMs:                   defaultRetryBackoffMs,
+		RetryBackoffMaxMs:                defaultRetryBackoffMaxMs,
+		FatalStatusCodes:                 strings.Split(defaultFatalStatusCodes, ","),
 	}
 
 	p.updateFromEnvVar()
@@ -130,6 +142,7 @@ func validateProviderConfiguration(p *ProviderConfiguration) error {
 
 // updateFromEnvVar is a utility to update configurations based on current environment variables
 func (cfg *ProviderConfiguration) updateFromEnvVar() {
+
 	portS := os.Getenv(flagdPortEnvironmentVariableName)
 	if portS != "" {
 		port, err := strconv.Atoi(portS)
@@ -159,17 +172,7 @@ func (cfg *ProviderConfiguration) updateFromEnvVar() {
 		cfg.CertPath = certificatePath
 	}
 
-	if maxCacheSizeS := os.Getenv(flagdMaxCacheSizeEnvironmentVariableName); maxCacheSizeS != "" {
-		maxCacheSizeFromEnv, err := strconv.Atoi(maxCacheSizeS)
-		if err != nil {
-			cfg.log.Error(err,
-				fmt.Sprintf("invalid env config for %s provided, using default value: %d",
-					flagdMaxCacheSizeEnvironmentVariableName, defaultMaxCacheSize,
-				))
-		} else {
-			cfg.MaxCacheSize = maxCacheSizeFromEnv
-		}
-	}
+	cfg.MaxCacheSize = getIntFromEnvVarOrDefault(flagdMaxCacheSizeEnvironmentVariableName, defaultMaxCacheSize, cfg.log)
 
 	if cacheValue := os.Getenv(flagdCacheEnvironmentVariableName); cacheValue != "" {
 		switch cache.Type(cacheValue) {
@@ -185,18 +188,8 @@ func (cfg *ProviderConfiguration) updateFromEnvVar() {
 		}
 	}
 
-	if maxEventStreamRetriesS := os.Getenv(
-		flagdMaxEventStreamRetriesEnvironmentVariableName); maxEventStreamRetriesS != "" {
-
-		maxEventStreamRetries, err := strconv.Atoi(maxEventStreamRetriesS)
-		if err != nil {
-			cfg.log.Error(err,
-				fmt.Sprintf("invalid env config for %s provided, using default value: %d",
-					flagdMaxEventStreamRetriesEnvironmentVariableName, defaultMaxEventStreamRetries))
-		} else {
-			cfg.EventStreamConnectionMaxAttempts = maxEventStreamRetries
-		}
-	}
+	cfg.EventStreamConnectionMaxAttempts = getIntFromEnvVarOrDefault(
+		flagdMaxEventStreamRetriesEnvironmentVariableName, defaultMaxEventStreamRetries, cfg.log)
 
 	if resolver := os.Getenv(flagdResolverEnvironmentVariableName); resolver != "" {
 		switch strings.ToLower(resolver) {
@@ -227,17 +220,34 @@ func (cfg *ProviderConfiguration) updateFromEnvVar() {
 	if targetUri := os.Getenv(flagdTargetUriEnvironmentVariableName); targetUri != "" {
 		cfg.TargetUri = targetUri
 	}
-	if gracePeriod := os.Getenv(flagdGracePeriodVariableName); gracePeriod != "" {
-		if seconds, err := strconv.Atoi(gracePeriod); err == nil {
-			cfg.RetryGracePeriod = seconds
+
+	cfg.RetryGracePeriod = getIntFromEnvVarOrDefault(flagdGracePeriodVariableName, defaultGracePeriod, cfg.log)
+	cfg.RetryBackoffMs = getIntFromEnvVarOrDefault(flagdRetryBackoffMsVariableName, defaultRetryBackoffMs, cfg.log)
+	cfg.RetryBackoffMaxMs = getIntFromEnvVarOrDefault(flagdRetryBackoffMaxMsVariableName, defaultRetryBackoffMaxMs, cfg.log)
+
+	if fatalStatusCodes := os.Getenv(flagdFatalStatusCodesVariableName); fatalStatusCodes != "" {
+		cfg.FatalStatusCodes = strings.Split(fatalStatusCodes, ",")
+	}
+}
+
+// Helper
+
+func getIntFromEnvVarOrDefault(envVarName string, defaultValue int, log logr.Logger) int {
+	if valueFromEnv := os.Getenv(envVarName); valueFromEnv != "" {
+		intValue, err := strconv.Atoi(valueFromEnv)
+		if err != nil {
+			log.Error(err,
+				fmt.Sprintf("invalid env config for %s provided, using default value: %d",
+					envVarName, defaultValue,
+				))
 		} else {
-			// Handle parsing error
-			cfg.log.Error(err, fmt.Sprintf("invalid grace period '%s'", gracePeriod))
+			return intValue
 		}
 	}
-
+	return defaultValue
 }
 
+
 // ProviderOptions
 
 type ProviderOption func(*ProviderConfiguration)
@@ -415,3 +425,25 @@ func WithRetryGracePeriod(gracePeriod int) ProviderOption {
 		p.RetryGracePeriod = gracePeriod
 	}
 }
+
+// WithRetryBackoffMs sets the initial backoff duration (in milliseconds) for retrying failed connections
+func WithRetryBackoffMs(retryBackoffMs int) ProviderOption {
+	return func(p *ProviderConfiguration) {
+		p.RetryBackoffMs = retryBackoffMs
+	}
+}
+
+// WithRetryBackoffMaxMs sets the maximum backoff duration (in milliseconds) for retrying failed connections
+func WithRetryBackoffMaxMs(retryBackoffMaxMs int) ProviderOption {
+	return func(p *ProviderConfiguration) {
+		p.RetryBackoffMaxMs = retryBackoffMaxMs
+	}
+}
+
+// WithFatalStatusCodes allows to set a list of gRPC status codes, which will cause streams to give up
+// and put the provider in a PROVIDER_FATAL state
+func WithFatalStatusCodes(fatalStatusCodes []string) ProviderOption {
+	return func(p *ProviderConfiguration) {
+		p.FatalStatusCodes = fatalStatusCodes
+	}
+}

providers/flagd/pkg/provider.go

@@ -74,6 +74,9 @@ func NewProvider(opts ...ProviderOption) (*Provider, error) {
 			CustomSyncProviderUri:   provider.providerConfiguration.CustomSyncProviderUri,
 			GrpcDialOptionsOverride: provider.providerConfiguration.GrpcDialOptionsOverride,
 			RetryGracePeriod:        provider.providerConfiguration.RetryGracePeriod,
+			RetryBackOffMs:          provider.providerConfiguration.RetryBackoffMs,
+			RetryBackOffMaxMs:       provider.providerConfiguration.RetryBackoffMaxMs,
+			FatalStatusCodes:        provider.providerConfiguration.FatalStatusCodes,
 		})
 	default:
 		service = process.NewInProcessService(process.Configuration{

providers/flagd/pkg/service/in_process/grpc_config.go

@@ -0,0 +1,67 @@
+package process
+
+import (
+	"encoding/json"
+	"strings"
+	"time"
+)
+
+const (
+	// Default timeouts and retry intervals
+	defaultKeepaliveTime    = 30 * time.Second
+	defaultKeepaliveTimeout = 5 * time.Second
+)
+
+type RetryPolicy struct {
+	MaxAttempts          int      `json:"MaxAttempts"`
+	InitialBackoff       string   `json:"InitialBackoff"`
+	MaxBackoff           string   `json:"MaxBackoff"`
+	BackoffMultiplier    float64  `json:"BackoffMultiplier"`
+	RetryableStatusCodes []string `json:"RetryableStatusCodes"`
+}
+
+func (g *Sync) buildRetryPolicy() string {
+	var policy = map[string]interface{}{
+		"methodConfig": []map[string]interface{}{
+			{
+				"name": []map[string]string{
+					{"service": "flagd.sync.v1.FlagSyncService"},
+				},
+				"retryPolicy": RetryPolicy{
+					MaxAttempts:          3,
+					InitialBackoff:		  (time.Duration(g.RetryBackOffMs) * time.Millisecond).String(),
+					MaxBackoff: 		  (time.Duration(g.RetryBackOffMaxMs) * time.Millisecond).String(),
+					BackoffMultiplier:    2.0,
+					RetryableStatusCodes: []string{"UNKNOWN","UNAVAILABLE"},
+				},
+			},
+		},
+	}
+	retryPolicyBytes, _ := json.Marshal(policy)
+	retryPolicy := string(retryPolicyBytes)
+
+	return retryPolicy
+}
+
+// Set of non-retryable gRPC status codes for faster lookup
+var nonRetryableCodes map[string]struct{}
+
+// initNonRetryableStatusCodesSet initializes the set of non-retryable gRPC status codes for quick lookup
+func (g *Sync) initNonRetryableStatusCodesSet()  {
+	nonRetryableCodes = make(map[string]struct{})
+	for _, code := range g.FatalStatusCodes {
+		normalized := toCamelCase(code)
+		nonRetryableCodes[normalized] = struct{}{}
+	}
+}
+
+// toCamelCase converts a SNAKE_CASE string to CamelCase
+func toCamelCase(s string) string {
+	parts := strings.Split(strings.ToLower(s), "_")
+	for i, part := range parts {
+		if len(part) > 0 {
+			parts[i] = strings.ToUpper(part[:1]) + part[1:]
+		}
+	}
+	return strings.Join(parts, "")
+}