fix(flagd): Add forbidden provider support and improve inprocess sync non-retry error handling (#756)

Signed-off-by: Alexandra Oberaigner <[email protected]>

Author: alexandraoberaigner

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

@@ -4,8 +4,8 @@ import (
 	"buf.build/gen/go/open-feature/flagd/grpc/go/flagd/sync/v1/syncv1grpc"
 	v1 "buf.build/gen/go/open-feature/flagd/protocolbuffers/go/flagd/sync/v1"
 	"context"
+	"encoding/json"
 	"fmt"
-	"github.com/goccy/go-json"
 	"github.com/open-feature/flagd/core/pkg/logger"
 	"github.com/open-feature/flagd/core/pkg/sync"
 	grpccredential "github.com/open-feature/flagd/core/pkg/sync/grpc/credentials"
@@ -14,6 +14,7 @@ import (
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/keepalive"
 	"google.golang.org/grpc/status"
+	"strings"
 	msync "sync"
 	"time"
 )
@@ -37,21 +38,8 @@ const (
 				"MaxBackoff": "5s",
 				"BackoffMultiplier": 2.0,
 				"RetryableStatusCodes": [
-				  "CANCELLED",
 				  "UNKNOWN",
-				  "INVALID_ARGUMENT",
-				  "NOT_FOUND",
-				  "ALREADY_EXISTS",
-				  "PERMISSION_DENIED",
-				  "RESOURCE_EXHAUSTED",
-				  "FAILED_PRECONDITION",
-				  "ABORTED",
-				  "OUT_OF_RANGE",
-				  "UNIMPLEMENTED",
-				  "INTERNAL",
-				  "UNAVAILABLE",
-				  "DATA_LOSS",
-				  "UNAUTHENTICATED"
+				  "UNAVAILABLE"
 				]
 			  }
 			}
@@ -61,7 +49,7 @@ const (
 	nonRetryableStatusCodes = `
 		[
 		  "PermissionDenied",
-		  "Unauthenticated",
+		  "Unauthenticated"
 		]
 	`
 )
@@ -90,6 +78,7 @@ type Sync struct {
 	Selector                string
 	URI                     string
 	MaxMsgSize              int
+	RetryGracePeriod		int
 
 	// Runtime state
 	client           FlagSyncServiceClient
@@ -104,7 +93,7 @@ type Sync struct {
 // Init initializes the gRPC connection and starts background monitoring
 func (g *Sync) Init(ctx context.Context) error {
 	g.Logger.Info(fmt.Sprintf("initializing gRPC client for %s", g.URI))
-	initNonRetryableStatusCodesSet()
+	g.initNonRetryableStatusCodesSet()
 
 	// Initialize channels
 	g.shutdownComplete = make(chan struct{})
@@ -174,13 +163,16 @@ func (g *Sync) buildDialOptions() ([]grpc.DialOption, error) {
 }
 
 // initNonRetryableStatusCodesSet initializes the set of non-retryable gRPC status codes for quick lookup
-func initNonRetryableStatusCodesSet()  {
+func (g *Sync) initNonRetryableStatusCodesSet()  {
 	var codes []string
 	nonRetryableCodes = make(map[string]struct{})
-	if err := json.Unmarshal([]byte(nonRetryableStatusCodes), &codes); err == nil {
+	trimmed := strings.TrimSpace(nonRetryableStatusCodes)
+	if err := json.Unmarshal([]byte(trimmed), &codes); err == nil {
 		for _, code := range codes {
 			nonRetryableCodes[code] = struct{}{}
 		}
+	} else {
+		g.Logger.Debug("parsing non-retryable status codes failed, retrying on all errors")
 	}
 }
 
@@ -243,11 +235,19 @@ func (g *Sync) Sync(ctx context.Context, dataSync chan<- sync.DataSync) error {
 			if ok {
 				codeStr := st.Code().String()
 				if _, found := nonRetryableCodes[codeStr]; found {
-					g.Logger.Error(fmt.Sprintf("sync cycle failed with non-retryable code: %v", codeStr))
-					return err
+					errStr := fmt.Sprintf("sync cycle failed with non-retryable status: %v, " +
+						"returning provider fatal.", codeStr)
+					g.Logger.Error(errStr)
+					return &of.ProviderInitError{
+						ErrorCode: of.ProviderFatalCode,
+						Message:   errStr,
+					}
 				}
 			}
 
+			// Backoff before retrying
+			time.Sleep(time.Duration(g.RetryGracePeriod))
+
 			g.Logger.Warn(fmt.Sprintf("sync cycle failed: %v, retrying...", err))
 			g.sendEvent(ctx, SyncEvent{event: of.ProviderError})
 

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

@@ -569,6 +569,7 @@ func createSyncProvider(cfg Configuration, log *logger.Logger) (isync.ISync, str
 		ProviderID:              cfg.ProviderID,
 		Selector:                cfg.Selector,
 		URI:                     uri,
+		RetryGracePeriod:        cfg.RetryGracePeriod,
 	}, uri
 }
 

providers/multi-provider/pkg/strategies/strategies.go

@@ -112,7 +112,7 @@ func setFlagMetadata(strategyUsed EvaluationStrategy, successProviderName string
 func cleanErrorMessage(msg string) string {
 	codeRegex := strings.Join([]string{
 		string(of.ProviderNotReadyCode),
-		// string(of.ProviderFatalCode), // TODO: not available until go-sdk 14
+		string(of.ProviderFatalCode),
 		string(of.FlagNotFoundCode),
 		string(of.ParseErrorCode),
 		string(of.TypeMismatchCode),

tests/flagd/testframework/provider_steps.go

@@ -27,6 +27,9 @@ func InitializeProviderSteps(ctx *godog.ScenarioContext) {
 	// Generic provider step definition - accepts any provider type including "stable"
 	ctx.Step(`^a (\w+) flagd provider$`,
 		withState1Arg((*TestState).createSpecializedFlagdProvider))
+
+	ctx.Step(`^the client is in (\w+) state$`,
+		withState1Arg((*TestState).assertClientState))
 }
 
 // State methods - these now expect context as first parameter after state
@@ -111,6 +114,13 @@ func (s *TestState) simulateConnectionLoss(ctx context.Context, seconds int) err
 	return s.Container.Restart(seconds)
 }
 
+func (s *TestState) assertClientState(ctx context.Context, state string) error {
+	if string(s.Client.State()) == strings.ToUpper(state) {
+		return nil
+	}
+	return fmt.Errorf("expected client state %s but got %s", state, s.Client.State())
+}
+
 // createSpecializedFlagdProvider creates specialized flagd providers based on type
 func (s *TestState) createSpecializedFlagdProvider(ctx context.Context, providerType string) error {
 	// Apply specialized configuration based on provider type
@@ -128,7 +138,7 @@ func (s *TestState) createSpecializedFlagdProvider(ctx context.Context, provider
 		return fmt.Errorf("failed to create instance for %s provider: %w", providerType, err)
 	}
 
-	if providerType != "unavailable" {
+	if providerType != "unavailable" && providerType != "forbidden" {
 		if s.ProviderType == RPC {
 			// Small delay to allow flagd server to fully load flags after connection
 			time.Sleep(50 * time.Millisecond)
@@ -150,6 +160,7 @@ func (s *TestState) applySpecializedConfig(providerType string) error {
 		return nil
 	case "unavailable":
 		return s.configureUnavailableProvider()
+	case "forbidden": return s.configureForbiddenProvider()
 	case "socket":
 		return s.configureSocketProvider()
 	case "ssl", "tls":
@@ -173,6 +184,12 @@ func (s *TestState) configureUnavailableProvider() error {
 	return nil
 }
 
+func (s *TestState) configureForbiddenProvider() error {
+	// Set an Envoy port which always responds with forbidden
+	s.addProviderOption("port", "Integer", "9212")
+	return nil
+}
+
 func (s *TestState) configureSocketProvider() error {
 	// Configure for unix socket connection
 	s.addProviderOption("socketPath", "String", "/tmp/flagd.sock")