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) }