return fatal only on initialization aka first stream sycle, use grpc numeric code representation for lookup of fatal status codes

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

Author: alexandraoberaigner

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

@@ -2,7 +2,8 @@ package process
 
 import (
 	"encoding/json"
-	"strings"
+	"fmt"
+	"google.golang.org/grpc/codes"
 	"time"
 )
 
@@ -29,10 +30,10 @@ func (g *Sync) buildRetryPolicy() string {
 				},
 				"retryPolicy": RetryPolicy{
 					MaxAttempts:          3,
-					InitialBackoff:		  (time.Duration(g.RetryBackOffMs) * time.Millisecond).String(),
-					MaxBackoff: 		  (time.Duration(g.RetryBackOffMaxMs) * time.Millisecond).String(),
+					InitialBackoff:       (time.Duration(g.RetryBackOffMs) * time.Millisecond).String(),
+					MaxBackoff:           (time.Duration(g.RetryBackOffMaxMs) * time.Millisecond).String(),
 					BackoffMultiplier:    2.0,
-					RetryableStatusCodes: []string{"UNKNOWN","UNAVAILABLE"},
+					RetryableStatusCodes: []string{"UNKNOWN", "UNAVAILABLE"},
 				},
 			},
 		},
@@ -44,24 +45,22 @@ func (g *Sync) buildRetryPolicy() string {
 }
 
 // Set of non-retryable gRPC status codes for faster lookup
-var nonRetryableCodes map[string]struct{}
+var nonRetryableCodes map[codes.Code]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{}{}
-	}
-}
+func (g *Sync) initNonRetryableStatusCodesSet() {
+	nonRetryableCodes = make(map[codes.Code]struct{})
+
+	for _, codeStr := range g.FatalStatusCodes {
+		// Wrap the string in quotes to match the expected JSON format
+		jsonStr := fmt.Sprintf(`"%s"`, codeStr)
 
-// 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:]
+		var code codes.Code
+		if err := code.UnmarshalJSON([]byte(jsonStr)); err != nil {
+			g.Logger.Warn(fmt.Sprintf("unknown status code: %s, error: %v", codeStr, err))
+			continue
 		}
+
+		nonRetryableCodes[code] = struct{}{}
 	}
-	return strings.Join(parts, "")
-}
+}

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

@@ -2,12 +2,13 @@ package process
 
 import (
 	"encoding/json"
-	"reflect"
+	"github.com/open-feature/flagd/core/pkg/logger"
+	"go.uber.org/zap"
+	"google.golang.org/grpc/codes"
 	"strings"
 	"testing"
 )
 
-
 func TestBuildRetryPolicy(t *testing.T) {
 	g := &Sync{
 		RetryBackOffMs:    100,
@@ -68,58 +69,71 @@ func TestBuildRetryPolicy(t *testing.T) {
 	}
 }
 
-type syncTestCase struct {
-	input    []string
-	expected map[string]struct{}
-}
-
-func TestInitNonRetryableStatusCodesSet(t *testing.T) {
-	testCases := []syncTestCase{
+func TestSync_initNonRetryableStatusCodesSet(t *testing.T) {
+	tests := []struct {
+		name             string
+		fatalStatusCodes []string
+		expectedCodes    []codes.Code
+		notExpectedCodes []codes.Code
+	}{
 		{
-			input:    []string{"PERMISSION_DENIED", "UNKNOWN"},
-			expected: map[string]struct{}{"PermissionDenied": {}, "Unknown": {}},
+			name:             "valid status codes",
+			fatalStatusCodes: []string{"UNAVAILABLE", "INTERNAL", "DEADLINE_EXCEEDED"},
+			expectedCodes:    []codes.Code{codes.Unavailable, codes.Internal, codes.DeadlineExceeded},
+			notExpectedCodes: []codes.Code{codes.OK, codes.Unknown},
 		},
 		{
-			input:    []string{"ALREADY_EXISTS"},
-			expected: map[string]struct{}{"AlreadyExists": {}},
+			name:             "empty array",
+			fatalStatusCodes: []string{},
+			expectedCodes:    []codes.Code{},
+			notExpectedCodes: []codes.Code{codes.Unavailable, codes.Internal},
 		},
 		{
-			input:    []string{},
-			expected: map[string]struct{}{},
+			name:             "invalid status codes",
+			fatalStatusCodes: []string{"INVALID_CODE", "UNKNOWN_STATUS"},
+			expectedCodes:    []codes.Code{},
+			notExpectedCodes: []codes.Code{codes.Unavailable, codes.Internal},
+		},
+		{
+			name:             "mixed valid and invalid codes",
+			fatalStatusCodes: []string{"UNAVAILABLE", "INVALID_CODE", "INTERNAL"},
+			expectedCodes:    []codes.Code{codes.Unavailable, codes.Internal},
+			notExpectedCodes: []codes.Code{codes.OK, codes.Unknown},
 		},
 	}
 
-	for _, tc := range testCases {
-		g := &Sync{FatalStatusCodes: tc.input}
-		nonRetryableCodes = nil // reset global
-		g.initNonRetryableStatusCodesSet()
-		if !reflect.DeepEqual(nonRetryableCodes, tc.expected) {
-			t.Errorf("input: %v, got: %v, want: %v", tc.input, nonRetryableCodes, tc.expected)
-		}
-	}
-}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Reset the global map before each test
+			nonRetryableCodes = nil
+
+			s := &Sync{
+				FatalStatusCodes: tt.fatalStatusCodes,
+				Logger: &logger.Logger{
+					Logger: zap.NewNop(),
+				},
+			}
+
+			s.initNonRetryableStatusCodesSet()
 
-func TestToCamelCase(t *testing.T) {
- testCases := []struct {
-  input    string
-  expected string
- }{
-  {"INVALID_ARGUMENT", "InvalidArgument"},
-  {"NOT_FOUND", "NotFound"},
-  {"ALREADY_EXISTS", "AlreadyExists"},
-  {"UNKNOWN", "Unknown"},
-  {"", ""},
-  {"SINGLE", "Single"},
-  {"MULTI_WORD_EXAMPLE", "MultiWordExample"},
-  {"_LEADING_UNDERSCORE", "LeadingUnderscore"},
-  {"TRAILING_UNDERSCORE_", "TrailingUnderscore"},
-  {"__DOUBLE__UNDERSCORES__", "DoubleUnderscores"},
- }
+			// Verify expected codes are present
+			for _, code := range tt.expectedCodes {
+				if _, exists := nonRetryableCodes[code]; !exists {
+					t.Errorf("expected code %v to be in nonRetryableCodes, but it was not found", code)
+				}
+			}
 
- for _, tc := range testCases {
-  got := toCamelCase(tc.input)
-  if got != tc.expected {
-   t.Errorf("toCamelCase(%q) = %q; want %q", tc.input, got, tc.expected)
-  }
- }
-}
+			// Verify not expected codes are absent
+			for _, code := range tt.notExpectedCodes {
+				if _, exists := nonRetryableCodes[code]; exists {
+					t.Errorf("did not expect code %v to be in nonRetryableCodes, but it was found", code)
+				}
+			}
+
+			// Verify the map size matches expected
+			if len(nonRetryableCodes) != len(tt.expectedCodes) {
+				t.Errorf("expected map size %d, got %d", len(tt.expectedCodes), len(nonRetryableCodes))
+			}
+		})
+	}
+}

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

@@ -38,10 +38,10 @@ type Sync struct {
 	Selector                string
 	URI                     string
 	MaxMsgSize              int
-	RetryGracePeriod		int
-	RetryBackOffMs			int
-	RetryBackOffMaxMs		int
-	FatalStatusCodes		[]string
+	RetryGracePeriod        int
+	RetryBackOffMs          int
+	RetryBackOffMaxMs       int
+	FatalStatusCodes        []string
 
 	// Runtime state
 	client           FlagSyncServiceClient
@@ -125,8 +125,6 @@ func (g *Sync) buildDialOptions() ([]grpc.DialOption, error) {
 	return dialOptions, nil
 }
 
-
-
 // ReSync performs a one-time fetch of all flags
 func (g *Sync) ReSync(ctx context.Context, dataSync chan<- sync.DataSync) error {
 	g.Logger.Debug("performing ReSync - fetching all flags")
@@ -160,8 +158,6 @@ func (g *Sync) IsReady() bool {
 func (g *Sync) Sync(ctx context.Context, dataSync chan<- sync.DataSync) error {
 	g.Logger.Info("starting continuous flag synchronization")
 
-	time.Sleep(500 * time.Millisecond)
-
 	// Ensure shutdown completion is signaled when THIS method exits
 	defer g.markShutdownComplete()
 
@@ -183,17 +179,18 @@ func (g *Sync) Sync(ctx context.Context, dataSync chan<- sync.DataSync) error {
 				return ctx.Err()
 			}
 
-			// Check if error is a gRPC status error and if code is retryable
-			st, ok := status.FromError(err)
-			if ok {
-				codeStr := st.Code().String()
-				if _, found := nonRetryableCodes[codeStr]; found {
-					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,
+			// check for non-retryable errors during initialization, if found return with FATAL
+			if !g.IsReady() {
+				st, ok := status.FromError(err)
+				if ok {
+					if _, found := nonRetryableCodes[st.Code()]; found {
+						errStr := fmt.Sprintf("first sync cycle failed with non-retryable status: %v, "+
+							"returning provider fatal.", st.Code().String())
+						g.Logger.Error(errStr)
+						return &of.ProviderInitError{
+							ErrorCode: of.ProviderFatalCode,
+							Message:   errStr,
+						}
 					}
 				}
 			}
@@ -236,12 +233,6 @@ func (g *Sync) performSyncCycle(ctx context.Context, dataSync chan<- sync.DataSy
 
 // handleFlagSync processes messages from the sync stream with proper context handling
 func (g *Sync) handleFlagSync(ctx context.Context, stream syncv1grpc.FlagSyncService_SyncFlagsClient, dataSync chan<- sync.DataSync) error {
-	// Mark as ready on first successful stream
-	g.initializer.Do(func() {
-		g.ready = true
-		g.Logger.Info("sync service is now ready")
-	})
-
 	// Create channels for stream communication
 	streamChan := make(chan *v1.SyncFlagsResponse, 1)
 	errChan := make(chan error, 1)
@@ -281,6 +272,11 @@ func (g *Sync) handleFlagSync(ctx context.Context, stream syncv1grpc.FlagSyncSer
 				return err
 			}
 
+			// Mark as ready on first successful stream
+			g.initializer.Do(func() {
+				g.ready = true
+				g.Logger.Info("sync service is now ready")
+			})
 		case err := <-errChan:
 			return fmt.Errorf("stream error: %w", err)