feat: report STALE reason when a sync source is disconnected

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

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

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"

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

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

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