From bd1ab3cdcb76f5035b3adf8d1a07e4f2691a1956 Mon Sep 17 00:00:00 2001 From: Alexandre Chakroun Date: Fri, 23 Oct 2026 15:17:38 +0200 Subject: [PATCH 1/2] feat: add option to close streams gracefully at server-side deadline Add --stream-deadline-graceful-close (FLAGD_STREAM_DEADLINE_GRACEFUL_CLOSE). When enabled, sync and event streams reaching the server-side deadline (--stream-deadline) end with an OK status instead of DEADLINE_EXCEEDED. Providers already treat a completed stream as a routine reconnect, so the periodic stream recycle no longer surfaces as an error in provider logs or gRPC client spans. Defaults to false, preserving current behavior. Refs: #1959 Co-Authored-By: Claude Signed-off-by: Alexandre Chakroun --- core/pkg/service/iservice.go | 1 + docs/reference/flagd-cli/flagd_start.md | 1 + flagd/cmd/start.go | 4 ++ flagd/pkg/runtime/from_config.go | 33 ++++++++-------- .../flag-evaluation/connect_service.go | 2 + .../flag-evaluation/flag_evaluator_v1.go | 6 +++ .../flag-evaluation/flag_evaluator_v2.go | 6 +++ flagd/pkg/service/flag-sync/handler.go | 8 +++- flagd/pkg/service/flag-sync/sync_service.go | 24 ++++++------ .../service/flag-sync/sync_service_test.go | 38 +++++++++++++------ 10 files changed, 84 insertions(+), 39 deletions(-) diff --git a/core/pkg/service/iservice.go b/core/pkg/service/iservice.go index 67f281195..512179556 100644 --- a/core/pkg/service/iservice.go +++ b/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 diff --git a/docs/reference/flagd-cli/flagd_start.md b/docs/reference/flagd-cli/flagd_start.md index 3339b8b74..811dff141 100644 --- a/docs/reference/flagd-cli/flagd_start.md +++ b/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. diff --git a/flagd/cmd/start.go b/flagd/cmd/start.go index 2749a19e4..bbb521e33 100644 --- a/flagd/cmd/start.go +++ b/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), diff --git a/flagd/pkg/runtime/from_config.go b/flagd/pkg/runtime/from_config.go index 931951ff2..fd34710ec 100644 --- a/flagd/pkg/runtime/from_config.go +++ b/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, }, diff --git a/flagd/pkg/service/flag-evaluation/connect_service.go b/flagd/pkg/service/flag-evaluation/connect_service.go index 6e49ae570..21a50110c 100644 --- a/flagd/pkg/service/flag-evaluation/connect_service.go +++ b/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)...) diff --git a/flagd/pkg/service/flag-evaluation/flag_evaluator_v1.go b/flagd/pkg/service/flag-evaluation/flag_evaluator_v1.go index a1391670c..d143d8fcb 100644 --- a/flagd/pkg/service/flag-evaluation/flag_evaluator_v1.go +++ b/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,6 +204,11 @@ func (s *FlagEvaluationService) EventStream( } case <-ctx.Done(): if errors.Is(ctx.Err(), context.DeadlineExceeded) { + if s.deadlineGraceful { + // end the stream with OK, providers treat a completed stream as a routine reconnect + s.logger.Debug(fmt.Sprintf("server-side deadline of %s exceeded, closing stream gracefully", s.deadline.String())) + return nil + } 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")) } diff --git a/flagd/pkg/service/flag-evaluation/flag_evaluator_v2.go b/flagd/pkg/service/flag-evaluation/flag_evaluator_v2.go index 1a3d68bd3..65e5c759f 100644 --- a/flagd/pkg/service/flag-evaluation/flag_evaluator_v2.go +++ b/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,6 +126,11 @@ func (s *FlagEvaluationServiceV2) EventStream( } case <-ctx.Done(): if errors.Is(ctx.Err(), context.DeadlineExceeded) { + if s.deadlineGraceful { + // end the stream with OK, providers treat a completed stream as a routine reconnect + s.logger.Debug(fmt.Sprintf("server-side deadline of %s exceeded, closing stream gracefully", s.deadline.String())) + return nil + } 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")) } diff --git a/flagd/pkg/service/flag-sync/handler.go b/flagd/pkg/service/flag-sync/handler.go index e6d976747..774862dde 100644 --- a/flagd/pkg/service/flag-sync/handler.go +++ b/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") diff --git a/flagd/pkg/service/flag-sync/sync_service.go b/flagd/pkg/service/flag-sync/sync_service.go index 666851287..a11aa6c37 100644 --- a/flagd/pkg/service/flag-sync/sync_service.go +++ b/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) diff --git a/flagd/pkg/service/flag-sync/sync_service_test.go b/flagd/pkg/service/flag-sync/sync_service_test.go index c9f58e068..40663421c 100644 --- a/flagd/pkg/service/flag-sync/sync_service_test.go +++ b/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 From 0980f738d0ffc4e6c4e7656be32789a759873722 Mon Sep 17 00:00:00 2001 From: Alexandre Chakroun Date: Fri, 23 Oct 2026 16:55:37 +0200 Subject: [PATCH 2/2] refactor: share event stream deadline handling between v1 and v2 Extract the server-side deadline result of the v1 and v2 event streams into deadlineExceededResult, removing the duplicated block, and cover both the graceful and non-graceful paths with a unit test. Co-Authored-By: Claude Signed-off-by: Alexandre Chakroun --- .../pkg/service/flag-evaluation/flag_evaluator.go | 11 +++++++++++ .../flag-evaluation/flag_evaluator_test.go | 15 +++++++++++++++ .../service/flag-evaluation/flag_evaluator_v1.go | 8 +------- .../service/flag-evaluation/flag_evaluator_v2.go | 8 +------- 4 files changed, 28 insertions(+), 14 deletions(-) diff --git a/flagd/pkg/service/flag-evaluation/flag_evaluator.go b/flagd/pkg/service/flag-evaluation/flag_evaluator.go index 1f21aced1..805a30fa2 100644 --- a/flagd/pkg/service/flag-evaluation/flag_evaluator.go +++ b/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) } diff --git a/flagd/pkg/service/flag-evaluation/flag_evaluator_test.go b/flagd/pkg/service/flag-evaluation/flag_evaluator_test.go index d7e45797c..3835a8696 100644 --- a/flagd/pkg/service/flag-evaluation/flag_evaluator_test.go +++ b/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)) + }) +} diff --git a/flagd/pkg/service/flag-evaluation/flag_evaluator_v1.go b/flagd/pkg/service/flag-evaluation/flag_evaluator_v1.go index d143d8fcb..795d41a01 100644 --- a/flagd/pkg/service/flag-evaluation/flag_evaluator_v1.go +++ b/flagd/pkg/service/flag-evaluation/flag_evaluator_v1.go @@ -204,13 +204,7 @@ func (s *FlagEvaluationService) EventStream( } case <-ctx.Done(): if errors.Is(ctx.Err(), context.DeadlineExceeded) { - if s.deadlineGraceful { - // end the stream with OK, providers treat a completed stream as a routine reconnect - s.logger.Debug(fmt.Sprintf("server-side deadline of %s exceeded, closing stream gracefully", s.deadline.String())) - return nil - } - 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 } diff --git a/flagd/pkg/service/flag-evaluation/flag_evaluator_v2.go b/flagd/pkg/service/flag-evaluation/flag_evaluator_v2.go index 65e5c759f..794b77c2f 100644 --- a/flagd/pkg/service/flag-evaluation/flag_evaluator_v2.go +++ b/flagd/pkg/service/flag-evaluation/flag_evaluator_v2.go @@ -126,13 +126,7 @@ func (s *FlagEvaluationServiceV2) EventStream( } case <-ctx.Done(): if errors.Is(ctx.Err(), context.DeadlineExceeded) { - if s.deadlineGraceful { - // end the stream with OK, providers treat a completed stream as a routine reconnect - s.logger.Debug(fmt.Sprintf("server-side deadline of %s exceeded, closing stream gracefully", s.deadline.String())) - return nil - } - 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 }