From c2279be0e0a2294a7b4f2aa0c95524d0ec70a69e Mon Sep 17 00:00:00 2001 From: Scott Holodak Date: Wed, 12 Aug 2026 16:37:22 -0400 Subject: [PATCH] feat: report STALE reason when a sync source is disconnected Closes #400. When a sync source goes offline, flagd keeps serving the flags it already holds -- deliberately, since last-known-good data beats failing evaluations open. Until now nothing distinguished those values from live ones: a flagd whose sync stream had dropped kept answering STATIC/TARGETING_MATCH indefinitely, /readyz stayed 200 (documented to latch on first successful sync and never change), and no metric exposed the condition. Behind a load balancer, one disconnected replica among healthy ones returns different values for the same flag on successive requests, with nothing to indicate which answer is current. This implements the behaviour agreed in #400, where STALE was added to the OpenFeature specification for exactly this case. - sync.DataSync gains a Stale field. It marks a connection-state notification rather than a flag payload; FlagData is ignored and the store is left untouched. - The gRPC sync emits one when its stream drops, both on the initial failure and on each failed re-establishment, and stops once a payload arrives. The send is best-effort against the context so a blocked runtime can never stall the reconnect loop. - store.SourceState records which sources are currently disconnected. A nil value reports nothing stale, so embedders that do not wire it up keep the previous behaviour exactly. - The runtime marks a source stale on such a notification and clears it on the next successful payload. It does not Emit: no flag data changed, and the sync protocol cannot convey staleness downstream. - The evaluator reports model.StaleReason for flags resolved from a disconnected source, wired via the new evaluator.WithSourceState option. STALE only replaces a successful resolution. Errors keep ERROR -- a failed evaluation has no value to be uncertain about -- and FALLBACK is left alone because it carries internal meaning translated in the API response. Because reasons already surface on feature_flag.flagd.result.reason, this makes a disconnected replica visible per pod with no new telemetry. Verified end to end against two flagd instances: STATIC while connected, STALE with the value still served once the source is killed, and STATIC again on reconnect. Signed-off-by: Scott Holodak --- core/pkg/evaluator/json.go | 26 ++++++ core/pkg/evaluator/stale_test.go | 133 +++++++++++++++++++++++++++ core/pkg/model/reason.go | 4 + core/pkg/store/source_state.go | 63 +++++++++++++ core/pkg/store/source_state_test.go | 72 +++++++++++++++ core/pkg/sync/grpc/grpc_sync.go | 13 +++ core/pkg/sync/grpc/grpc_sync_test.go | 55 ++++++++--- core/pkg/sync/isync.go | 8 ++ flagd/pkg/runtime/from_config.go | 10 +- flagd/pkg/runtime/runtime.go | 19 ++++ 10 files changed, 386 insertions(+), 17 deletions(-) create mode 100644 core/pkg/evaluator/stale_test.go create mode 100644 core/pkg/store/source_state.go create mode 100644 core/pkg/store/source_state_test.go diff --git a/core/pkg/evaluator/json.go b/core/pkg/evaluator/json.go index 2d4a4f9e2..69e26b1d6 100644 --- a/core/pkg/evaluator/json.go +++ b/core/pkg/evaluator/json.go @@ -140,6 +140,18 @@ type Resolver struct { store store.IStore Logger *logger.Logger tracer trace.Tracer + // sourceState is optional; when nil, no flag is ever reported stale and + // behaviour is identical to releases before stale reporting existed. + sourceState *store.SourceState +} + +// WithSourceState wires per-source connection state into the evaluator, so that +// flags resolved from a sync source which is currently disconnected are reported +// with model.StaleReason instead of their usual reason. +func WithSourceState(s *store.SourceState) JSONEvaluatorOption { + return func(je *JSON) { + je.Resolver.sourceState = s + } } func NewResolver(store store.IStore, logger *logger.Logger, jsonEvalTracer trace.Tracer) Resolver { @@ -346,6 +358,20 @@ func (je *Resolver) evaluateVariant(ctx context.Context, reqID string, flagKey s } } + // A flag resolved from a disconnected sync source may no longer match the + // source of truth. flagd deliberately keeps serving last-known-good data + // rather than failing the evaluation, so the uncertainty is surfaced in the + // reason instead. STALE only ever replaces a successful resolution: errors + // keep their own reason, and FALLBACK is left alone because it carries + // internal meaning that is translated in the API response. + if je.sourceState.IsStale(flag.Source) { + defer func() { + if err == nil && reason != model.ErrorReason && reason != model.FallbackReason { + reason = model.StaleReason + } + }() + } + if flag.State == Disabled { je.Logger.DebugWithID(reqID, fmt.Sprintf("requested flag is disabled: %s", flagKey)) return "", nil, model.DisabledReason, metadata, nil diff --git a/core/pkg/evaluator/stale_test.go b/core/pkg/evaluator/stale_test.go new file mode 100644 index 000000000..1657efb1f --- /dev/null +++ b/core/pkg/evaluator/stale_test.go @@ -0,0 +1,133 @@ +//nolint:wrapcheck +package evaluator_test + +import ( + "context" + "testing" + + flagdEvaluator "github.com/open-feature/flagd/core/pkg/evaluator" + "github.com/open-feature/flagd/core/pkg/logger" + "github.com/open-feature/flagd/core/pkg/model" + "github.com/open-feature/flagd/core/pkg/store" + "github.com/open-feature/flagd/core/pkg/sync" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const staleSource = "testSource" + +func staleEvaluator(t *testing.T, state *store.SourceState) *flagdEvaluator.JSON { + t.Helper() + + e := flagdEvaluator.NewJSON( + logger.NewLogger(nil, false), + store.NewFlags(), + flagdEvaluator.WithSourceState(state), + ) + require.NoError(t, e.SetState(sync.DataSync{FlagData: flagConfig, Source: staleSource})) + + return e +} + +// Flags are still served while their source is disconnected -- serving +// last-known-good data beats failing evaluations open -- but the reason tells +// the caller the value may no longer match the source of truth. +func TestStale_ReplacesSuccessfulReasons(t *testing.T) { + tests := []struct { + name string + flagKey string + evalCtx map[string]interface{} + freshReason string + expectedVal bool + }{ + { + name: "static resolution", + flagKey: StaticBoolFlag, + freshReason: model.StaticReason, + expectedVal: StaticBoolValue, + }, + { + name: "targeting match", + flagKey: DynamicBoolFlag, + evalCtx: map[string]interface{}{ColorProp: ColorValue}, + freshReason: model.TargetingMatchReason, + expectedVal: StaticBoolValue, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + state := store.NewSourceState() + e := staleEvaluator(t, state) + + // baseline: a connected source keeps its usual reason + val, _, reason, _, err := e.ResolveBooleanValue(context.TODO(), "default", test.flagKey, test.evalCtx) + require.NoError(t, err) + assert.Equal(t, test.freshReason, reason) + assert.Equal(t, test.expectedVal, val) + + // the source drops; the value is unchanged but now reported stale + state.SetStale(staleSource, true) + + val, _, reason, _, err = e.ResolveBooleanValue(context.TODO(), "default", test.flagKey, test.evalCtx) + require.NoError(t, err) + assert.Equal(t, model.StaleReason, reason, "a disconnected source must resolve as STALE") + assert.Equal(t, test.expectedVal, val, "the last-known-good value must still be served") + + // the source recovers + state.SetStale(staleSource, false) + + _, _, reason, _, err = e.ResolveBooleanValue(context.TODO(), "default", test.flagKey, test.evalCtx) + require.NoError(t, err) + assert.Equal(t, test.freshReason, reason, "reconnecting must restore the original reason") + }) + } +} + +// STALE reports uncertainty about a value that was resolved. An evaluation that +// failed has no value to be uncertain about, so its error reason must survive. +func TestStale_DoesNotMaskErrors(t *testing.T) { + tests := []struct { + name string + flagKey string + errorCode string + }{ + {"missing flag", MissingFlag, model.FlagNotFoundErrorCode}, + {"type mismatch", StaticObjectFlag, model.TypeMismatchErrorCode}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + state := store.NewSourceState() + state.SetStale(staleSource, true) + e := staleEvaluator(t, state) + + _, _, reason, _, err := e.ResolveBooleanValue(context.TODO(), "default", test.flagKey, nil) + + assert.EqualError(t, err, test.errorCode) + assert.Equal(t, model.ErrorReason, reason, "errors must keep ERROR, not be rewritten to STALE") + }) + } +} + +func TestStale_OnlyAffectsTheDisconnectedSource(t *testing.T) { + state := store.NewSourceState() + e := staleEvaluator(t, state) + + state.SetStale("some-other-source", true) + + _, _, reason, _, err := e.ResolveBooleanValue(context.TODO(), "default", StaticBoolFlag, nil) + require.NoError(t, err) + assert.Equal(t, model.StaticReason, reason, "an unrelated stale source must not affect this flag") +} + +// Source tracking is opt-in. Without it flagd must behave exactly as it did +// before stale reporting existed. +func TestStale_NoSourceStateConfigured(t *testing.T) { + e := flagdEvaluator.NewJSON(logger.NewLogger(nil, false), store.NewFlags()) + require.NoError(t, e.SetState(sync.DataSync{FlagData: flagConfig, Source: staleSource})) + + _, _, reason, _, err := e.ResolveBooleanValue(context.TODO(), "default", StaticBoolFlag, nil) + require.NoError(t, err) + assert.Equal(t, model.StaticReason, reason) +} diff --git a/core/pkg/model/reason.go b/core/pkg/model/reason.go index 96f19fe30..79ca61f39 100644 --- a/core/pkg/model/reason.go +++ b/core/pkg/model/reason.go @@ -10,6 +10,10 @@ const ( UnknownReason = "UNKNOWN" ErrorReason = "ERROR" StaticReason = "STATIC" + // StaleReason indicates the flag was resolved from a store whose sync source is + // currently disconnected, so the value may no longer reflect the source of truth. + // See https://openfeature.dev/specification/types#resolution-details + StaleReason = "STALE" // only used internally if no default value could be determined // will be translated to DefaultReason in the API response FallbackReason = "FALLBACK" diff --git a/core/pkg/store/source_state.go b/core/pkg/store/source_state.go new file mode 100644 index 000000000..7faa190ac --- /dev/null +++ b/core/pkg/store/source_state.go @@ -0,0 +1,63 @@ +package store + +import "sync" + +// SourceState tracks, per sync source, whether that source is currently +// disconnected. Flags already held in the store remain servable while a source +// is down -- flagd deliberately keeps serving last-known-good data rather than +// failing evaluations -- but consumers deserve to know the data may be out of +// date. Evaluations resolved from a disconnected source are reported with +// model.StaleReason. +// +// It is written by the runtime (driven by sync.DataSync payloads) and read on +// the evaluation hot path, so reads are cheap and lock-free-ish via RWMutex. +// The zero value is not usable; construct with NewSourceState. +type SourceState struct { + mu sync.RWMutex + stale map[string]bool +} + +func NewSourceState() *SourceState { + return &SourceState{stale: map[string]bool{}} +} + +// SetStale records whether the given source is currently disconnected. +func (s *SourceState) SetStale(source string, stale bool) { + if s == nil { + return + } + s.mu.Lock() + defer s.mu.Unlock() + if stale { + s.stale[source] = true + return + } + delete(s.stale, source) +} + +// IsStale reports whether the given source is currently disconnected. A nil +// receiver reports false so that callers which never wire up source tracking +// (tests, embedders) behave exactly as before. +func (s *SourceState) IsStale(source string) bool { + if s == nil { + return false + } + s.mu.RLock() + defer s.mu.RUnlock() + return s.stale[source] +} + +// StaleSources returns the sources currently marked disconnected. Intended for +// diagnostics and tests; order is not guaranteed. +func (s *SourceState) StaleSources() []string { + if s == nil { + return nil + } + s.mu.RLock() + defer s.mu.RUnlock() + sources := make([]string, 0, len(s.stale)) + for source := range s.stale { + sources = append(sources, source) + } + return sources +} diff --git a/core/pkg/store/source_state_test.go b/core/pkg/store/source_state_test.go new file mode 100644 index 000000000..bbe5952d3 --- /dev/null +++ b/core/pkg/store/source_state_test.go @@ -0,0 +1,72 @@ +package store + +import "testing" + +func TestSourceState_DefaultsToNotStale(t *testing.T) { + s := NewSourceState() + + if s.IsStale("grpc://example:8015") { + t.Fatal("a source with no recorded state must not be reported stale") + } +} + +func TestSourceState_SetAndClear(t *testing.T) { + const source = "grpc://example:8015" + s := NewSourceState() + + s.SetStale(source, true) + if !s.IsStale(source) { + t.Fatal("expected source to be stale after SetStale(true)") + } + + s.SetStale(source, false) + if s.IsStale(source) { + t.Fatal("expected source to be fresh after SetStale(false)") + } + if got := len(s.StaleSources()); got != 0 { + t.Fatalf("expected no stale sources retained, got %d", got) + } +} + +func TestSourceState_IsolatesSources(t *testing.T) { + s := NewSourceState() + s.SetStale("a", true) + + if !s.IsStale("a") { + t.Fatal("expected source a to be stale") + } + if s.IsStale("b") { + t.Fatal("marking source a stale must not affect source b") + } +} + +// A nil SourceState is the zero-configuration case: embedders and tests that +// never wire up source tracking must see exactly the pre-existing behaviour +// rather than a panic. +func TestSourceState_NilReceiverIsSafe(t *testing.T) { + var s *SourceState + + s.SetStale("a", true) + if s.IsStale("a") { + t.Fatal("a nil SourceState must never report a source stale") + } + if s.StaleSources() != nil { + t.Fatal("a nil SourceState must return no stale sources") + } +} + +func TestSourceState_ConcurrentAccess(t *testing.T) { + s := NewSourceState() + done := make(chan struct{}) + + go func() { + for i := 0; i < 1000; i++ { + s.SetStale("a", i%2 == 0) + } + close(done) + }() + for i := 0; i < 1000; i++ { + _ = s.IsStale("a") + } + <-done +} diff --git a/core/pkg/sync/grpc/grpc_sync.go b/core/pkg/sync/grpc/grpc_sync.go index d11e45beb..2a535cc87 100644 --- a/core/pkg/sync/grpc/grpc_sync.go +++ b/core/pkg/sync/grpc/grpc_sync.go @@ -147,6 +147,7 @@ func (g *Sync) Sync(ctx context.Context, dataSync chan<- sync.DataSync) error { } g.Logger.Warn(fmt.Sprintf("error with stream listener: %s", err.Error())) + g.notifyStale(ctx, dataSync) // retry connection establishment for { @@ -159,11 +160,23 @@ func (g *Sync) Sync(ctx context.Context, dataSync chan<- sync.DataSync) error { err = g.handleFlagSync(syncClient, dataSync) if err != nil { g.Logger.Warn(fmt.Sprintf("error with stream listener: %s", err.Error())) + g.notifyStale(ctx, dataSync) continue } } } +// notifyStale tells the runtime that this source is disconnected, so evaluations +// served from its flags can be reported as stale. Flags stay in the store; only +// the reported reason changes. The send is best-effort: a blocked or cancelled +// runtime must never stall the reconnection loop. +func (g *Sync) notifyStale(ctx context.Context, dataSync chan<- sync.DataSync) { + select { + case dataSync <- sync.DataSync{Source: g.URI, Stale: true}: + case <-ctx.Done(): + } +} + // connectWithRetry is a helper that performs exponential back off after retrying connection attempts periodically until // a successful connection is established. Caller must not expect an error. Hence, errors are handled, logged // internally. However, if the provided context is done, method exit with a non-ok state which must be verified by the diff --git a/core/pkg/sync/grpc/grpc_sync_test.go b/core/pkg/sync/grpc/grpc_sync_test.go index 693a8410e..22a288253 100644 --- a/core/pkg/sync/grpc/grpc_sync_test.go +++ b/core/pkg/sync/grpc/grpc_sync_test.go @@ -358,9 +358,14 @@ func Test_ReSyncTests(t *testing.T) { } } - // channel must be empty - if len(syncChan) != 0 { - t.Errorf("Data sync channel must be empty after all test syncs. But received non empty: %d", len(syncChan)) + // No further flag payloads may arrive. A stream ending also emits a stale + // notification, which carries no flag data and is expected here. + for len(syncChan) > 0 { + leftover := <-syncChan + if !leftover.Stale { + t.Errorf("Data sync channel must hold no flag payloads after all test syncs, but received: %q", + leftover.FlagData) + } } } } @@ -503,9 +508,14 @@ func Test_StreamListener(t *testing.T) { } } - // channel must be empty - if len(syncChan) != 0 { - t.Errorf("Data sync channel must be empty after all test syncs. But received non empty: %d", len(syncChan)) + // No further flag payloads may arrive. A stream ending also emits a stale + // notification, which carries no flag data and is expected here. + for len(syncChan) > 0 { + leftover := <-syncChan + if !leftover.Stale { + t.Errorf("Data sync channel must hold no flag payloads after all test syncs, but received: %q", + leftover.FlagData) + } } } } @@ -726,7 +736,7 @@ func Test_SyncRetry(t *testing.T) { break case data := <-syncChan: if data.FlagData != emptyFlagData { - t.Errorf("sync start error: %s", err.Error()) + t.Errorf("expected flag data %q, but got %q", emptyFlagData, data.FlagData) } } @@ -740,16 +750,31 @@ func Test_SyncRetry(t *testing.T) { // Restart the server go serve(&bServer) - // validate connection re-establishment - select { - case <-tCtx.Done(): - cancelFunc() - t.Error("timeout waiting for conditions to fulfil") - case data := <-syncChan: - if data.FlagData != emptyFlagData { - t.Errorf("sync start error: %s", err.Error()) + // Validate connection re-establishment. Losing the stream now also emits a + // stale notification (carrying no flag data) ahead of the re-delivered + // payload, so drain messages until the payload arrives. + sawStale := false + for settled := false; !settled; { + select { + case <-tCtx.Done(): + cancelFunc() + t.Error("timeout waiting for conditions to fulfil") + settled = true + case data := <-syncChan: + if data.Stale { + sawStale = true + continue + } + if data.FlagData != emptyFlagData { + t.Errorf("expected flag data %q, but got %q", emptyFlagData, data.FlagData) + } + settled = true } } + + if !sawStale { + t.Error("expected a stale notification to be emitted when the sync stream dropped") + } } // Mock implementations diff --git a/core/pkg/sync/isync.go b/core/pkg/sync/isync.go index dc1e1de52..0517c30d9 100644 --- a/core/pkg/sync/isync.go +++ b/core/pkg/sync/isync.go @@ -38,6 +38,14 @@ type DataSync struct { // explicitly opted-in per source via SourceConfig.IncrementalUpdates. // EXPERIMENTAL: this option may change or be removed in a future release. IncrementalUpdates bool + + // Stale marks this message as a connection-state notification rather than a + // flag payload: the source is currently disconnected, so the flags already + // held for it may no longer match the source of truth. FlagData is ignored + // for such messages and the store is left untouched -- flagd keeps serving + // last-known-good data -- but evaluations resolved from this source are + // reported with model.StaleReason until a subsequent payload arrives. + Stale bool } // SourceConfig is configuration option for flagd. This maps to startup parameter sources diff --git a/flagd/pkg/runtime/from_config.go b/flagd/pkg/runtime/from_config.go index 777a1af98..6097d17e8 100644 --- a/flagd/pkg/runtime/from_config.go +++ b/flagd/pkg/runtime/from_config.go @@ -93,6 +93,11 @@ func FromConfig(logger *logger.Logger, version string, config Config) (*Runtime, sources = append(sources, provider.URI) } + // tracks which sync sources are currently disconnected, so evaluations served + // from their (retained) flags can be reported with model.StaleReason. + // Declared before the `store` variable below, which shadows the package name. + sourceState := store.NewSourceState() + // build flag store, collect flag sources & fill sources details store, err := store.NewStore(logger, sources) if err != nil { @@ -100,7 +105,7 @@ func FromConfig(logger *logger.Logger, version string, config Config) (*Runtime, } // derive evaluator - jsonEvaluator := evaluator.NewJSON(logger, store) + jsonEvaluator := evaluator.NewJSON(logger, store, evaluator.WithSourceState(sourceState)) // derive services @@ -182,7 +187,8 @@ func FromConfig(logger *logger.Logger, version string, config Config) (*Runtime, MaxRequestBodyBytes: config.MaxRequestBodyBytes, MaxRequestHeaderBytes: config.MaxRequestHeaderBytes, }, - Syncs: iSyncs, + Syncs: iSyncs, + SourceState: sourceState, }, nil } diff --git a/flagd/pkg/runtime/runtime.go b/flagd/pkg/runtime/runtime.go index df03f2663..62bfc070d 100644 --- a/flagd/pkg/runtime/runtime.go +++ b/flagd/pkg/runtime/runtime.go @@ -12,6 +12,7 @@ import ( "github.com/open-feature/flagd/core/pkg/evaluator" "github.com/open-feature/flagd/core/pkg/logger" "github.com/open-feature/flagd/core/pkg/service" + "github.com/open-feature/flagd/core/pkg/store" "github.com/open-feature/flagd/core/pkg/sync" "github.com/open-feature/flagd/flagd/pkg/service/flag-evaluation/ofrep" flagsync "github.com/open-feature/flagd/flagd/pkg/service/flag-sync" @@ -26,6 +27,9 @@ type Runtime struct { EvaluationService service.IFlagEvaluationService ServiceConfig service.Configuration Syncs []sync.ISync + // SourceState is optional; when nil, sync disconnections are logged but no + // evaluation is reported stale. + SourceState *store.SourceState mu msync.Mutex } @@ -127,10 +131,25 @@ func (r *Runtime) updateAndEmit(payload sync.DataSync) { r.mu.Lock() defer r.mu.Unlock() + if payload.Stale { + // Connection-state notification, not flag data. The source's flags stay in + // the store on purpose -- continuing to serve last-known-good values beats + // failing evaluations open -- but they are now reported with + // model.StaleReason so callers can tell the data may be out of date. + // No Emit: nothing about the flag payload changed, and the sync protocol + // has no way to convey staleness to downstream clients. + r.SourceState.SetStale(payload.Source, true) + r.Logger.Warn(fmt.Sprintf( + "sync source %s is disconnected; its flags are still served but now reported as stale", payload.Source)) + return + } + err := r.Evaluator.SetState(payload) if err != nil { r.Logger.Error(fmt.Sprintf("error setting state: %v", err)) return } + // A successful payload means the source is reachable again. + r.SourceState.SetStale(payload.Source, false) r.SyncService.Emit(payload.Source) }