feat: add option to close streams gracefully at server-side deadline

core/pkg/service/iservice.go

@@ -36,6 +36,7 @@ type Configuration struct {
 	ContextValues              map[string]any
 	HeaderToContextKeyMappings map[string]string
 	StreamDeadline             time.Duration
+	StreamDeadlineGraceful     bool
 	MaxRequestHeaderBytes      int64
 	MaxRequestBodyBytes        int64
 

docs/reference/flagd-cli/flagd_start.md

@@ -38,6 +38,7 @@ flagd start [flags]
   -d, --socket-path string                   Flagd unix socket path. With grpc the evaluations service will become available on this address. With http(s) the grpc-gateway proxy will use this address internally.
   -s, --sources string                       JSON representation of an array of SourceConfig objects. Required fields: uri (string) and provider (string). Optional source-specific fields are also available, see https://flagd.dev/reference/sync-configuration/#source-configuration
       --stream-deadline duration             Set a server-side deadline for flagd sync and event streams (default 0, means no deadline).
+      --stream-deadline-graceful-close       Close streams reaching the server-side deadline with an OK status instead of DEADLINE_EXCEEDED, so providers reconnect without reporting an error. Defaults to false.
       --sync-http-enabled                    Serve the flag configuration document over HTTP at /v1/flags on the sync port. Defaults to true. (default true)
   -g, --sync-port int32                      Flag sync service port, serving gRPC, gRPC-Web and Connect (default 8015)
   -e, --sync-socket-path string              Flagd sync service socket path. With grpc the sync service will be available on this address.

flagd/cmd/start.go

@@ -45,6 +45,7 @@ const (
 	contextValueFlagName       = "context-value"
 	headerToContextKeyFlagName = "context-from-header"
 	streamDeadlineFlagName     = "stream-deadline"
+	streamDeadlineGracefulFlag = "stream-deadline-graceful-close"
 	maxRequestBodyFlagName     = "max-request-body"
 	maxRequestHeaderFlagName   = "max-request-header"
 
@@ -105,6 +106,7 @@ func init() {
 	flags.StringToStringP(headerToContextKeyFlagName, "H", map[string]string{}, "add key-value pairs to map "+
 		"header values to context values, where key is Header name, value is context key")
 	flags.Duration(streamDeadlineFlagName, 0, "Set a server-side deadline for flagd sync and event streams (default 0, means no deadline).")
+	flags.Bool(streamDeadlineGracefulFlag, false, "Close streams reaching the server-side deadline with an OK status instead of DEADLINE_EXCEEDED, so providers reconnect without reporting an error. Defaults to false.")
 	flags.Bool(disableSyncMetadata, false, "Disables the getMetadata endpoint of the sync service. Defaults to false, but will default to true in later versions.")
 	flags.Int64P(maxRequestBodyFlagName, "B", 1_000_000, "Maximum allowed request body size in bytes. Requests exceeding this are rejected with HTTP 413 (OFREP) or 429 (connect). Set to 0 to disable. WARNING: disabling this limit may allow memory exhaustion from oversized requests.")
 	flags.Int64P(maxRequestHeaderFlagName, "R", 1_000_000, "Maximum allowed request header size in bytes. Requests exceeding this are rejected with HTTP 431. Set to 0 to use Go's built-in default (1 MiB). WARNING: setting a very large or zero value may allow memory exhaustion from oversized headers.")
@@ -139,6 +141,7 @@ func bindFlags(flags *pflag.FlagSet) {
 	_ = viper.BindPFlag(contextValueFlagName, flags.Lookup(contextValueFlagName))
 	_ = viper.BindPFlag(headerToContextKeyFlagName, flags.Lookup(headerToContextKeyFlagName))
 	_ = viper.BindPFlag(streamDeadlineFlagName, flags.Lookup(streamDeadlineFlagName))
+	_ = viper.BindPFlag(streamDeadlineGracefulFlag, flags.Lookup(streamDeadlineGracefulFlag))
 	_ = viper.BindPFlag(disableSyncMetadata, flags.Lookup(disableSyncMetadata))
 	_ = viper.BindPFlag(maxRequestBodyFlagName, flags.Lookup(maxRequestBodyFlagName))
 	_ = viper.BindPFlag(maxRequestHeaderFlagName, flags.Lookup(maxRequestHeaderFlagName))
@@ -239,6 +242,7 @@ var startCmd = &cobra.Command{
 			SyncServiceHTTPEnabled: viper.GetBool(syncHTTPFlagName),
 			SyncServiceSocketPath:  viper.GetString(syncSocketPathFlagName),
 			StreamDeadline:         viper.GetDuration(streamDeadlineFlagName),
+			StreamDeadlineGraceful: viper.GetBool(streamDeadlineGracefulFlag),
 			DisableSyncMetadata:    viper.GetBool(disableSyncMetadata),
 			//nolint:staticcheck // SA1019 no-ops, read so FromConfig can warn when they are set
 			KeepAliveMinTime: viper.GetDuration(keepAliveMinTimeFlagName),

flagd/pkg/runtime/from_config.go

@@ -44,6 +44,7 @@ type Config struct {
 	SyncServiceHTTPEnabled bool
 	SyncServiceSocketPath  string
 	StreamDeadline         time.Duration
+	StreamDeadlineGraceful bool
 	DisableSyncMetadata    bool
 
 	// Deprecated: no-op, retained so existing configurations keep working. See warnOnNoOpConfig.
@@ -168,21 +169,22 @@ func FromConfig(logger *logger.Logger, version string, config Config) (*Runtime,
 
 	// flag sync service
 	flagSyncService, err := flagsync.NewSyncService(flagsync.SvcConfigurations{
-		Logger:              logger.WithFields(zap.String("component", "FlagSyncService")),
-		Port:                config.SyncServicePort,
-		Sources:             sources,
-		Store:               store,
-		ContextValues:       config.ContextValues,
-		KeyPath:             config.ServiceKeyPath,
-		CertPath:            config.ServiceCertPath,
-		SocketPath:          config.SyncServiceSocketPath,
-		HTTPEnabled:         config.SyncServiceHTTPEnabled,
-		ServiceName:         svcName,
-		CORS:                config.CORS,
-		Options:             options,
-		StreamDeadline:      config.StreamDeadline,
-		DisableSyncMetadata: config.DisableSyncMetadata,
-		MetricsRecorder:     recorder,
+		Logger:                 logger.WithFields(zap.String("component", "FlagSyncService")),
+		Port:                   config.SyncServicePort,
+		Sources:                sources,
+		Store:                  store,
+		ContextValues:          config.ContextValues,
+		KeyPath:                config.ServiceKeyPath,
+		CertPath:               config.ServiceCertPath,
+		SocketPath:             config.SyncServiceSocketPath,
+		HTTPEnabled:            config.SyncServiceHTTPEnabled,
+		ServiceName:            svcName,
+		CORS:                   config.CORS,
+		Options:                options,
+		StreamDeadline:         config.StreamDeadline,
+		StreamDeadlineGraceful: config.StreamDeadlineGraceful,
+		DisableSyncMetadata:    config.DisableSyncMetadata,
+		MetricsRecorder:        recorder,
 
 		MaxRequestHeaderBytes: config.MaxRequestHeaderBytes,
 		MaxRequestBodyBytes:   config.MaxRequestBodyBytes,
@@ -219,6 +221,7 @@ func FromConfig(logger *logger.Logger, version string, config Config) (*Runtime,
 			ContextValues:              config.ContextValues,
 			HeaderToContextKeyMappings: config.HeaderToContextKeyMappings,
 			StreamDeadline:             config.StreamDeadline,
+			StreamDeadlineGraceful:     config.StreamDeadlineGraceful,
 			MaxRequestBodyBytes:        config.MaxRequestBodyBytes,
 			MaxRequestHeaderBytes:      config.MaxRequestHeaderBytes,
 		},

flagd/pkg/service/flag-evaluation/connect_service.go

@@ -160,6 +160,7 @@ func (s *ConnectService) setupServer(svcConf service.Configuration) (net.Listene
 		svcConf.HeaderToContextKeyMappings,
 		svcConf.StreamDeadline,
 	)
+	v1Fes.deadlineGraceful = svcConf.StreamDeadlineGraceful
 
 	_, v1Handler := evaluationV1.NewServiceHandler(v1Fes, append(svcConf.Options, marshalOpts)...)
 
@@ -173,6 +174,7 @@ func (s *ConnectService) setupServer(svcConf service.Configuration) (net.Listene
 		svcConf.HeaderToContextKeyMappings,
 		svcConf.StreamDeadline,
 	)
+	v2Fes.deadlineGraceful = svcConf.StreamDeadlineGraceful
 
 	_, v2Handler := evaluationV2.NewServiceHandler(v2Fes, append(svcConf.Options, marshalOpts)...)
 

flagd/pkg/service/flag-evaluation/flag_evaluator.go

@@ -358,6 +358,17 @@ func (s *OldFlagEvaluationService) ResolveObject(
 	return res, err
 }
 
+// deadlineExceededResult is the result of an event stream reaching its server-side deadline:
+// OK when graceful, so providers reconnect as for a completed stream, DEADLINE_EXCEEDED otherwise
+func deadlineExceededResult(log *logger.Logger, deadline time.Duration, graceful bool) error {
+	if graceful {
+		log.Debug(fmt.Sprintf("server-side deadline of %s exceeded, closing stream gracefully", deadline.String()))
+		return nil
+	}
+	log.Debug(fmt.Sprintf("server-side deadline of %s exceeded, exiting stream request with grpc error code 4", deadline.String()))
+	return connect.NewError(connect.CodeDeadlineExceeded, fmt.Errorf("%s", "stream closed due to server-side timeout"))
+}
+
 func mergeContexts(reqCtx, configFlagsCtx map[string]any, headers http.Header, headerToContextKeyMappings map[string]string) map[string]any {
 	return MergeContextsAndHeaders(reqCtx, configFlagsCtx, headers, headerToContextKeyMappings)
 }

flagd/pkg/service/flag-evaluation/flag_evaluator_test.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"errors"
 	"testing"
+	"time"
 
 	schemaV1 "buf.build/gen/go/open-feature/flagd/protocolbuffers/go/schema/v1"
 	"connectrpc.com/connect"
@@ -1112,3 +1113,17 @@ func assertInvalidSelectorConnect(t *testing.T, cases []invalidSelectorCase) {
 		})
 	}
 }
+
+func TestDeadlineExceededResult(t *testing.T) {
+	log := logger.NewLogger(nil, false)
+
+	t.Run("graceful closes the stream with OK", func(t *testing.T) {
+		require.NoError(t, deadlineExceededResult(log, time.Minute, true))
+	})
+
+	t.Run("non-graceful returns DEADLINE_EXCEEDED", func(t *testing.T) {
+		err := deadlineExceededResult(log, time.Minute, false)
+		require.Error(t, err)
+		require.Equal(t, connect.CodeDeadlineExceeded, connect.CodeOf(err))
+	})
+}

flagd/pkg/service/flag-evaluation/flag_evaluator_v1.go

@@ -30,6 +30,7 @@ type FlagEvaluationService struct {
 	contextValues              map[string]any
 	headerToContextKeyMappings map[string]string
 	deadline                   time.Duration
+	deadlineGraceful           bool
 }
 
 // NewFlagEvaluationService creates a FlagEvaluationService with provided parameters
@@ -203,8 +204,7 @@ func (s *FlagEvaluationService) EventStream(
 			}
 		case <-ctx.Done():
 			if errors.Is(ctx.Err(), context.DeadlineExceeded) {
-				s.logger.Debug(fmt.Sprintf("server-side deadline of %s exceeded, exiting stream request with grpc error code 4", s.deadline.String()))
-				return connect.NewError(connect.CodeDeadlineExceeded, fmt.Errorf("%s", "stream closed due to server-side timeout"))
+				return deadlineExceededResult(s.logger, s.deadline, s.deadlineGraceful)
 			}
 			return nil
 		}

flagd/pkg/service/flag-evaluation/flag_evaluator_v2.go

@@ -32,6 +32,7 @@ type FlagEvaluationServiceV2 struct {
 	contextValues              map[string]any
 	headerToContextKeyMappings map[string]string
 	deadline                   time.Duration
+	deadlineGraceful           bool
 }
 
 // NewFlagEvaluationServiceV2 creates a FlagEvaluationServiceV2 with provided parameters
@@ -125,8 +126,7 @@ func (s *FlagEvaluationServiceV2) EventStream(
 			}
 		case <-ctx.Done():
 			if errors.Is(ctx.Err(), context.DeadlineExceeded) {
-				s.logger.Debug(fmt.Sprintf("server-side deadline of %s exceeded, exiting stream request with grpc error code 4", s.deadline.String()))
-				return connect.NewError(connect.CodeDeadlineExceeded, fmt.Errorf("%s", "stream closed due to server-side timeout"))
+				return deadlineExceededResult(s.logger, s.deadline, s.deadlineGraceful)
 			}
 			return nil
 		}

flagd/pkg/service/flag-sync/handler.go

@@ -26,6 +26,7 @@ type syncHandler struct {
 	log                 *logger.Logger
 	contextValues       map[string]any
 	deadline            time.Duration
+	deadlineGraceful    bool
 	disableSyncMetadata bool
 	metricsRecorder     telemetry.IMetricsRecorder
 }
@@ -115,8 +116,13 @@ func (s syncHandler) syncFlags(
 			}
 		case <-ctx.Done():
 			if errors.Is(ctx.Err(), context.DeadlineExceeded) {
-				s.log.Debug(fmt.Sprintf("server-side deadline of %s exceeded, exiting stream request with grpc error code 4", s.deadline.String()))
 				exitReason = "deadline_exceeded"
+				if s.deadlineGraceful {
+					// end the stream with OK, providers treat a completed stream as a routine reconnect
+					s.log.Debug(fmt.Sprintf("server-side deadline of %s exceeded, closing stream gracefully", s.deadline.String()))
+					return nil
+				}
+				s.log.Debug(fmt.Sprintf("server-side deadline of %s exceeded, exiting stream request with grpc error code 4", s.deadline.String()))
 				return connect.NewError(connect.CodeDeadlineExceeded, errors.New("stream closed due to server-side timeout"))
 			}
 			s.log.Debug("context complete and exiting stream request")

flagd/pkg/service/flag-sync/sync_service.go

@@ -30,17 +30,18 @@ type ISyncService interface {
 }
 
 type SvcConfigurations struct {
-	Logger              *logger.Logger
-	Port                uint16
-	Sources             []string
-	Store               store.IStore
-	ContextValues       map[string]any
-	CertPath            string
-	KeyPath             string
-	SocketPath          string
-	StreamDeadline      time.Duration
-	DisableSyncMetadata bool
-	MetricsRecorder     telemetry.IMetricsRecorder
+	Logger                 *logger.Logger
+	Port                   uint16
+	Sources                []string
+	Store                  store.IStore
+	ContextValues          map[string]any
+	CertPath               string
+	KeyPath                string
+	SocketPath             string
+	StreamDeadline         time.Duration
+	StreamDeadlineGraceful bool
+	DisableSyncMetadata    bool
+	MetricsRecorder        telemetry.IMetricsRecorder
 
 	HTTPEnabled           bool
 	ServiceName           string
@@ -96,6 +97,7 @@ func NewSyncService(cfg SvcConfigurations) (*Service, error) {
 		log:                 l,
 		contextValues:       cfg.ContextValues,
 		deadline:            cfg.StreamDeadline,
+		deadlineGraceful:    cfg.StreamDeadlineGraceful,
 		disableSyncMetadata: cfg.DisableSyncMetadata,
 		metricsRecorder:     cfg.MetricsRecorder,
 	}, flagsHandler)

flagd/pkg/service/flag-sync/sync_service_test.go

@@ -2,7 +2,9 @@ package sync
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"io"
 	"log"
 	"testing"
 	"time"
@@ -55,6 +57,7 @@ func TestSyncServiceEndToEnd(t *testing.T) {
 					tc.socketPath,
 					ctx,
 					0,
+					false,
 					disableSyncMetadata,
 				)
 
@@ -166,9 +169,11 @@ func TestSyncServiceDeadlineEndToEnd(t *testing.T) {
 	testCases := []struct {
 		title    string
 		deadline time.Duration
+		graceful bool
 	}{
 		{title: "without deadline", deadline: 0},
 		{title: "with deadline", deadline: 2 * time.Second},
+		{title: "with graceful deadline", deadline: 2 * time.Second, graceful: true},
 	}
 
 	for _, tc := range testCases {
@@ -184,7 +189,7 @@ func TestSyncServiceDeadlineEndToEnd(t *testing.T) {
 			ctx, cancelFunc := context.WithCancel(context.Background())
 			defer cancelFunc()
 
-			_, doneChan, err := createAndStartSyncService(port, sources, flagStore, certPath, keyPath, socketPath, ctx, tc.deadline, false)
+			_, doneChan, err := createAndStartSyncService(port, sources, flagStore, certPath, keyPath, socketPath, ctx, tc.deadline, tc.graceful, false)
 			if err != nil {
 				t.Fatal("error creating sync service")
 			}
@@ -232,10 +237,17 @@ func TestSyncServiceDeadlineEndToEnd(t *testing.T) {
 					// received data, continuing..
 					break
 				case err := <-errorChan:
+					if errors.Is(err, io.EOF) {
+						if !tc.graceful {
+							t.Fatal("stream closed with OK status even though graceful deadline was not configured.")
+						}
+						// expected graceful close due to deadline
+						return
+					}
 					st, _ := status.FromError(err)
 					if st.Code() == codes.DeadlineExceeded {
-						if tc.deadline == 0 {
-							t.Fatal("ran into deadline exceeded error even though no deadline was configured.")
+						if tc.deadline == 0 || tc.graceful {
+							t.Fatal("ran into deadline exceeded error even though no (non-graceful) deadline was configured.")
 						}
 						// expected error due to deadline
 						return
@@ -261,18 +273,20 @@ func createAndStartSyncService(
 	socketPath string,
 	ctx context.Context,
 	deadline time.Duration,
+	deadlineGraceful bool,
 	disableSyncMetadata bool,
 ) (*Service, chan interface{}, error) {
 	service, err := NewSyncService(SvcConfigurations{
-		Logger:              logger.NewLogger(nil, false),
-		Port:                uint16(port),
-		Sources:             sources,
-		Store:               store,
-		CertPath:            certPath,
-		KeyPath:             keyPath,
-		SocketPath:          socketPath,
-		StreamDeadline:      deadline,
-		DisableSyncMetadata: disableSyncMetadata,
+		Logger:                 logger.NewLogger(nil, false),
+		Port:                   uint16(port),
+		Sources:                sources,
+		Store:                  store,
+		CertPath:               certPath,
+		KeyPath:                keyPath,
+		SocketPath:             socketPath,
+		StreamDeadline:         deadline,
+		StreamDeadlineGraceful: deadlineGraceful,
+		DisableSyncMetadata:    disableSyncMetadata,
 	})
 	if err != nil {
 		return nil, nil, err