fix(store): deduplicate flags across flagSetIds in incremental update mode

Add a membership table to track which flagSetIds reference which flags,
enabling per-flagSetId queries without storing duplicate flag entries.
Flags are stored once per (key, source) in the flags table while the
membership table records (flagSetId, key, source) triples.

Key behaviors:
- Get iterates all membership entries for a (flagSetId, key) and selects
  the highest-priority flag, consistent with GetAll/collectViaMembership.
- updateIncremental upserts flag content when a (key, source) already
  exists at equal or lower priority, so content changes (targeting rules,
  variants) are not silently dropped.
- Metadata always includes flagSetId even when the original metadata is
  nil, ensuring consistent serialization for clients.
- Watch registers scoped per-(key, source) watch channels from memdb
  instead of watching the entire flags table, avoiding O(N_watchers x
  N_updates) spurious wake-ups.

Author: erenatas

core/pkg/store/bench_memory_test.go

@@ -0,0 +1,133 @@
+package store
+
+import (
+	"encoding/json"
+	"fmt"
+	"runtime"
+	"testing"
+
+	"github.com/open-feature/flagd/core/pkg/logger"
+	"github.com/open-feature/flagd/core/pkg/model"
+)
+
+// TestMemoryUsage simulates the production workload
+// It measures actual heap usage for both dedup (incremental) and no-dedup (full snapshot) modes.
+func TestMemoryUsage(t *testing.T) {
+	const (
+		numProjects     = 1000
+		commonFlagCount = 1000
+		totalUnique     = 3000
+		source          = "grpc://features-api:443"
+	)
+
+	// Build a realistic flag with variants, targeting, and metadata (~800B each)
+	makeFlag := func(key string) model.Flag {
+		return model.Flag{
+			Key:            key,
+			State:          "ENABLED",
+			DefaultVariant: "off",
+			Variants: map[string]any{
+				"on":  true,
+				"off": false,
+			},
+			Targeting: json.RawMessage(`{"if":[{"in":["@example.com",{"var":"email"}]},"on","off"]}`),
+		}
+	}
+
+	// Pre-generate all unique flags
+	allFlags := make([]model.Flag, totalUnique)
+	for i := range totalUnique {
+		allFlags[i] = makeFlag(fmt.Sprintf("flag-%04d", i))
+	}
+
+	commonFlags := allFlags[:commonFlagCount]
+
+	// Distribute remaining flags across projects
+	uniquePerProject := (totalUnique - commonFlagCount) / numProjects
+	if uniquePerProject < 1 {
+		uniquePerProject = 1
+	}
+
+	// Build per-project flag sets
+	type projectPayload struct {
+		flagSetId string
+		flags     []model.Flag
+	}
+	projects := make([]projectPayload, numProjects)
+	uniqueIdx := commonFlagCount
+	for i := range numProjects {
+		fsi := fmt.Sprintf("project-%04d", i)
+		// Each project gets all common flags + a slice of the unique ones
+		pFlags := make([]model.Flag, 0, commonFlagCount+uniquePerProject+1)
+		pFlags = append(pFlags, commonFlags...)
+		end := uniqueIdx + uniquePerProject
+		if i < (totalUnique-commonFlagCount)%numProjects {
+			end++ // distribute remainder across the first N projects
+		}
+		if end > totalUnique {
+			end = totalUnique
+		}
+		for ; uniqueIdx < end; uniqueIdx++ {
+			pFlags = append(pFlags, allFlags[uniqueIdx])
+		}
+		projects[i] = projectPayload{flagSetId: fsi, flags: pFlags}
+	}
+
+	measure := func() uint64 {
+		runtime.GC()
+		runtime.GC()
+		var m runtime.MemStats
+		runtime.ReadMemStats(&m)
+		return m.HeapAlloc
+	}
+
+	t.Run("with_dedup_incremental", func(t *testing.T) {
+		s, err := NewStore(logger.NewLogger(nil, false), []string{source})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		before := measure()
+
+		for _, p := range projects {
+			s.Update(source, p.flags, model.Metadata{"flagSetId": p.flagSetId}, true)
+		}
+
+		after := measure()
+		heapMB := float64(after-before) / (1024 * 1024)
+		t.Logf("DEDUP:    heap delta = %.1f MB (before=%d, after=%d)", heapMB, before, after)
+
+		// Sanity: verify flag count
+		allF, _, _ := s.GetAll(nil, nil)
+		t.Logf("          flags in store: %d (expected ~%d unique)", len(allF), totalUnique)
+	})
+
+	t.Run("without_dedup_full_snapshot", func(t *testing.T) {
+		s, err := NewStore(logger.NewLogger(nil, false), []string{source})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		before := measure()
+
+		// Full snapshot: each project replaces the whole source, so we simulate
+		// what the store would look like if all projects' flags were loaded at once
+		// with unique flagSetIds (no dedup). Load all at once since full-snapshot
+		// replaces per source.
+		var allWithFSI []model.Flag
+		for _, p := range projects {
+			for _, f := range p.flags {
+				f.Metadata = model.Metadata{"flagSetId": p.flagSetId}
+				allWithFSI = append(allWithFSI, f)
+			}
+		}
+		s.Update(source, allWithFSI, nil, false)
+
+		after := measure()
+		heapMB := float64(after-before) / (1024 * 1024)
+		t.Logf("NO DEDUP: heap delta = %.1f MB (before=%d, after=%d)", heapMB, before, after)
+
+		allF, _, _ := s.GetAll(nil, nil)
+		t.Logf("          flags in store: %d (expected ~%d with duplicates)", len(allF), numProjects*(commonFlagCount+uniquePerProject))
+	})
+}

core/pkg/store/query.go

@@ -24,6 +24,10 @@ const flagSetIdSourceCompoundIndex = flagSetIdIndex + "+" + sourceIndex
 const keySourceCompoundIndex = keyIndex + "+" + sourceIndex
 const flagSetIdKeySourceCompoundIndex = flagSetIdIndex + "+" + keyIndex + "+" + sourceIndex
 
+// membership table for incremental update deduplication
+const membershipTable = "membership"
+const membershipFlagSetIdKeyIndex = flagSetIdIndex + "+" + keyIndex
+
 // flagSetId defaults to a UUID generated at startup to make our queries consistent
 // any flag without a "flagSetId" is assigned this one; it's never exposed externally
 var nilFlagSetId = uuid.New().String()
@@ -83,6 +87,15 @@ func (s *Selector) IsEmpty() bool {
 	return s == nil || len(s.indexMap) == 0
 }
 
+// HasFlagSetId returns the flagSetId value and true if the selector includes a flagSetId constraint.
+func (s *Selector) HasFlagSetId() (string, bool) {
+	if s == nil || s.indexMap == nil {
+		return "", false
+	}
+	v, ok := s.indexMap[flagSetIdIndex]
+	return v, ok && v != ""
+}
+
 // ToQuery converts the Selector map to an indexId and constraints for querying the Store.
 // For a given index, a specific order and number of constraints are required.
 // Both the indexId and constraints are generated based on the keys present in the selector's internal map.