diff --git a/.changeset/flag-definition-cache-provider.md b/.changeset/flag-definition-cache-provider.md new file mode 100644 index 0000000..71880b3 --- /dev/null +++ b/.changeset/flag-definition-cache-provider.md @@ -0,0 +1,5 @@ +--- +"posthog-go": minor +--- + +Add the experimental `FlagDefinitionCacheProvider` interface and the `Config.FlagDefinitionCacheProvider` option, for sharing local evaluation flag definitions across SDK instances. See [Distributed environments](https://posthog.com/docs/feature-flags/local-evaluation/distributed-environments). diff --git a/api/public-api.txt b/api/public-api.txt index e18fd5d..c98bbcc 100644 --- a/api/public-api.txt +++ b/api/public-api.txt @@ -314,6 +314,8 @@ type Config struct { NextFeatureFlagsPollingTick func() time.Duration + FlagDefinitionCacheProvider FlagDefinitionCacheProvider + HistoricalMigration bool Transport http.RoundTripper @@ -624,6 +626,24 @@ type Filter struct { VariantLookupTable []FlagVariantMeta `json:"-"` } +type FlagDefinitionCacheData struct { + Flags []FeatureFlag `json:"flags"` + GroupTypeMapping map[string]string `json:"group_type_mapping"` + Cohorts map[string]PropertyGroup `json:"cohorts"` + MinimalFlagCalledEvents bool `json:"minimal_flag_called_events"` +} + +type FlagDefinitionCacheProvider interface { + GetFlagDefinitions(ctx context.Context) (*FlagDefinitionCacheData, error) + + ShouldFetchFlagDefinitions(ctx context.Context) (bool, error) + + OnFlagDefinitionsReceived(ctx context.Context, data FlagDefinitionCacheData) error + + Shutdown(ctx context.Context) error +} + + type FlagDetail struct { Key string `json:"key"` Enabled bool `json:"enabled"` diff --git a/config.go b/config.go index 040010f..1651b7b 100644 --- a/config.go +++ b/config.go @@ -126,6 +126,14 @@ type Config struct { // polling delay. When set, it overrides DefaultFeatureFlagsPollingInterval. NextFeatureFlagsPollingTick func() time.Duration + // FlagDefinitionCacheProvider shares local evaluation flag definitions with other + // SDK instances through an external cache. Only used when SecretKey is configured. + // See https://posthog.com/docs/feature-flags/local-evaluation/distributed-environments + // for guidance. + // + // EXPERIMENTAL: this API may change in a minor version bump. + FlagDefinitionCacheProvider FlagDefinitionCacheProvider + // HistoricalMigration marks captured batches as historical migration traffic. // See https://posthog.com/docs/migrate for migration guidance. HistoricalMigration bool diff --git a/examples/README.md b/examples/README.md index 7bb6351..fab5c08 100644 --- a/examples/README.md +++ b/examples/README.md @@ -31,6 +31,7 @@ This will run: - Feature flags example - Capture events example - Capture events with feature flag options example +- Feature flags definition cache example ### Prerequisites diff --git a/examples/flag_definition_cache.go b/examples/flag_definition_cache.go new file mode 100644 index 0000000..06a2609 --- /dev/null +++ b/examples/flag_definition_cache.go @@ -0,0 +1,205 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "strconv" + "strings" + "time" + + "github.com/google/uuid" + "github.com/posthog/posthog-go" +) + +// This example implements posthog.FlagDefinitionCacheProvider with a shared directory, +// so it runs without extra dependencies. A real deployment shares definitions across +// hosts, so use Redis or something like it there. + +const ( + // Must be longer than the polling interval. + flagCacheLockTTL = 30 * time.Second + flagCachePollInterval = 3 * time.Second + flagCacheInstances = 2 +) + +// FileFlagCache shares flag definitions between processes on one machine through a +// directory: one file holds the definitions, another is the leader election lock. +type FileFlagCache struct { + dir string + name string + instanceID string +} + +// NewFileFlagCache creates a provider for the given cache directory. Instances that +// share definitions pass the same directory and name. +func NewFileFlagCache(dir, name string) (*FileFlagCache, error) { + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, err + } + return &FileFlagCache{dir: dir, name: name, instanceID: uuid.NewString()}, nil +} + +func (c *FileFlagCache) cachePath() string { return filepath.Join(c.dir, c.name+".json") } +func (c *FileFlagCache) lockPath() string { return filepath.Join(c.dir, c.name+".lock") } + +// ShouldFetchFlagDefinitions elects a single instance to poll PostHog by holding a lock +// that expires. +// +// Read-then-write is not atomic on a filesystem, so two instances starting at the same +// moment can both take the lock. Redis with a Lua script does this atomically. +func (c *FileFlagCache) ShouldFetchFlagDefinitions(_ context.Context) (bool, error) { + holder, expiry, err := c.readLock() + if err != nil { + return false, err + } + + heldByAnother := holder != "" && holder != c.instanceID + if heldByAnother && time.Now().Before(expiry) { + return false, nil + } + + return true, c.writeLock() +} + +func (c *FileFlagCache) GetFlagDefinitions(_ context.Context) (*posthog.FlagDefinitionCacheData, error) { + contents, err := os.ReadFile(c.cachePath()) + if errors.Is(err, os.ErrNotExist) { + return nil, nil + } else if err != nil { + return nil, err + } + + var data posthog.FlagDefinitionCacheData + if err := json.Unmarshal(contents, &data); err != nil { + return nil, err + } + return &data, nil +} + +// OnFlagDefinitionsReceived publishes definitions, writing to a temporary file and +// renaming it so that a reader never observes a half-written payload. +func (c *FileFlagCache) OnFlagDefinitionsReceived(_ context.Context, data posthog.FlagDefinitionCacheData) error { + encoded, err := json.Marshal(data) + if err != nil { + return err + } + + tmp, err := os.CreateTemp(c.dir, c.name+".*.tmp") + if err != nil { + return err + } + defer os.Remove(tmp.Name()) + + if _, err := tmp.Write(encoded); err != nil { + tmp.Close() + return err + } + if err := tmp.Close(); err != nil { + return err + } + + fmt.Printf(" [%s] published %d flag definitions to the shared cache\n", c.shortID(), len(data.Flags)) + return os.Rename(tmp.Name(), c.cachePath()) +} + +func (c *FileFlagCache) Shutdown(_ context.Context) error { + holder, _, err := c.readLock() + if err != nil || holder != c.instanceID { + return err + } + if err := os.Remove(c.lockPath()); err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + return nil +} + +func (c *FileFlagCache) readLock() (holder string, expiry time.Time, err error) { + contents, err := os.ReadFile(c.lockPath()) + if errors.Is(err, os.ErrNotExist) { + return "", time.Time{}, nil + } else if err != nil { + return "", time.Time{}, err + } + + parts := strings.Fields(string(contents)) + if len(parts) != 2 { + return "", time.Time{}, nil + } + nanos, err := strconv.ParseInt(parts[1], 10, 64) + if err != nil { + return "", time.Time{}, nil + } + return parts[0], time.Unix(0, nanos), nil +} + +func (c *FileFlagCache) writeLock() error { + expiry := time.Now().Add(flagCacheLockTTL).UnixNano() + return os.WriteFile(c.lockPath(), []byte(fmt.Sprintf("%s %d", c.instanceID, expiry)), 0o644) +} + +func (c *FileFlagCache) shortID() string { return c.instanceID[:8] } + +// TestFlagDefinitionCache runs several clients that share one flag definition cache. +func TestFlagDefinitionCache(projectAPIKey, secretKey, endpoint string) { + dir, err := os.MkdirTemp("", "posthog-flag-cache") + if err != nil { + fmt.Printf("❌ Could not create the cache directory: %v\n", err) + return + } + defer os.RemoveAll(dir) + + fmt.Printf("%d instances sharing the cache in %s\n\n", flagCacheInstances, dir) + + clients := make([]posthog.Client, 0, flagCacheInstances) + caches := make([]*FileFlagCache, 0, flagCacheInstances) + + for i := 0; i < flagCacheInstances; i++ { + cache, err := NewFileFlagCache(dir, "my-service-production") + if err != nil { + fmt.Printf("❌ Could not create the cache provider: %v\n", err) + return + } + + client, err := posthog.NewWithConfig(projectAPIKey, posthog.Config{ + Endpoint: endpoint, + SecretKey: secretKey, + DefaultFeatureFlagsPollingInterval: flagCachePollInterval, + FlagDefinitionCacheProvider: cache, + }) + if err != nil { + fmt.Printf("❌ Could not create the client: %v\n", err) + return + } + + clients = append(clients, client) + caches = append(caches, cache) + fmt.Printf(" instance %s started\n", cache.shortID()) + } + + defer func() { + for i, client := range clients { + if err := client.Close(); err != nil { + fmt.Printf(" [%s] close: %v\n", caches[i].shortID(), err) + } + } + }() + + time.Sleep(2 * flagCachePollInterval) + + fmt.Println("\nEvaluating flags on every instance, entirely from the shared definitions:") + for i, client := range clients { + flags, err := client.GetAllFlags(posthog.FeatureFlagPayloadNoKey{ + DistinctId: "distinct-id-of-your-user", + OnlyEvaluateLocally: true, + }) + if err != nil { + fmt.Printf(" [%s] %v\n", caches[i].shortID(), err) + continue + } + fmt.Printf(" [%s] evaluated %d flags locally\n", caches[i].shortID(), len(flags)) + } +} diff --git a/examples/main.go b/examples/main.go index 93a894e..ea7c182 100644 --- a/examples/main.go +++ b/examples/main.go @@ -91,8 +91,9 @@ func showMenu() { fmt.Println("4. Feature flag with SendFeatureFlagsOptions examples") fmt.Println("5. Flag dependencies examples") fmt.Println("6. ETag polling test (continuous, Ctrl+C to stop)") - fmt.Println("7. Run all examples (except ETag polling)") - fmt.Println("8. Exit") + fmt.Println("7. Distributed flag definition cache examples") + fmt.Println("8. Run all examples (except ETag polling)") + fmt.Println("9. Exit") } func runBasicCaptureExamples() { @@ -125,6 +126,11 @@ func runETagPollingExample() { TestETagPolling(projectAPIKey, secretKey, endpoint) } +func runFlagDefinitionCacheExample() { + printExampleSection("DISTRIBUTED FLAG DEFINITION CACHE EXAMPLES") + TestFlagDefinitionCache(projectAPIKey, secretKey, endpoint) +} + func printExampleSection(title string) { fmt.Println("\n" + strings.Repeat("=", 60)) fmt.Println(title) @@ -150,6 +156,9 @@ func runAllExamples() { fmt.Printf("\n%s FLAG DEPENDENCIES %s\n", strings.Repeat("🔸", 20), strings.Repeat("🔸", 20)) TestFlagDependencies(projectAPIKey, secretKey, endpoint) + + fmt.Printf("\n%s FLAG DEFINITION CACHE %s\n", strings.Repeat("🔸", 19), strings.Repeat("🔸", 19)) + TestFlagDefinitionCache(projectAPIKey, secretKey, endpoint) } func isInteractive() bool { @@ -170,7 +179,7 @@ func main() { for { showMenu() - choice := promptForInput("\nEnter your choice (1-8): ") + choice := promptForInput("\nEnter your choice (1-9): ") switch choice { case "1": @@ -188,12 +197,14 @@ func main() { // ETag polling runs continuously, so exit after it returns return case "7": - runAllExamples() + runFlagDefinitionCacheExample() case "8": + runAllExamples() + case "9": fmt.Println("👋 Goodbye!") return default: - fmt.Println("❌ Invalid choice. Please select 1-8.") + fmt.Println("❌ Invalid choice. Please select 1-9.") continue } diff --git a/featureflags.go b/featureflags.go index c06c28e..93e85a9 100644 --- a/featureflags.go +++ b/featureflags.go @@ -84,6 +84,7 @@ type FeatureFlagsPoller struct { // After the request the channel get closed. firstFeatureFlagRequestFinished chan bool shutdown chan bool + shutdownDone chan bool forceReload chan bool // state holds all flag-related data using atomic pointer for lock-free reads @@ -95,12 +96,13 @@ type FeatureFlagsPoller struct { // Logger receives poller warnings and errors. Logger Logger // Endpoint is the PostHog API host used by the poller. - Endpoint string - http http.Client - nextPollTick func() time.Duration - flagTimeout time.Duration - decider decider - disableGeoIP bool + Endpoint string + http http.Client + nextPollTick func() time.Duration + flagTimeout time.Duration + decider decider + disableGeoIP bool + cacheProvider FlagDefinitionCacheProvider } // FeatureFlag is a feature flag definition returned by the local evaluation endpoint. @@ -461,6 +463,7 @@ func newFeatureFlagsPoller( flagTimeout time.Duration, decider decider, disableGeoIP bool, + cacheProvider FlagDefinitionCacheProvider, ) (*FeatureFlagsPoller, error) { localEvaluationEndpoint := "/flags/definitions" localEvalURL, err := url.Parse(endpoint + localEvaluationEndpoint) @@ -475,6 +478,7 @@ func newFeatureFlagsPoller( poller := FeatureFlagsPoller{ firstFeatureFlagRequestFinished: make(chan bool), shutdown: make(chan bool), + shutdownDone: make(chan bool), forceReload: make(chan bool), personalApiKey: personalApiKey, projectApiKey: projectApiKey, @@ -486,6 +490,7 @@ func newFeatureFlagsPoller( flagTimeout: flagTimeout, decider: decider, disableGeoIP: disableGeoIP, + cacheProvider: cacheProvider, } go poller.run() @@ -493,6 +498,8 @@ func newFeatureFlagsPoller( } func (poller *FeatureFlagsPoller) run() { + defer close(poller.shutdownDone) + poller.fetchNewFeatureFlags() close(poller.firstFeatureFlagRequestFinished) @@ -512,10 +519,121 @@ func (poller *FeatureFlagsPoller) run() { } } -// fetchNewFeatureFlags fetches the latest feature flag definitions from the PostHog API -// These are used for local evaluation of feature flags and should not be confused with -// the feature flags fetched from the flags API. +// fetchNewFeatureFlags refreshes the local feature flag definitions used for local +// evaluation. These should not be confused with the feature flags fetched from the +// flags API. func (poller *FeatureFlagsPoller) fetchNewFeatureFlags() { + if poller.cacheProvider == nil { + poller.fetchFlagDefinitions(false) + return + } + + shouldFetch, err := poller.cacheProvider.ShouldFetchFlagDefinitions(context.Background()) + if err != nil { + poller.Logger.Errorf("[FEATURE FLAGS] Cache provider ShouldFetchFlagDefinitions failed, fetching from the API: %s", err) + shouldFetch = true + } + + if !shouldFetch { + if poller.loadFlagDefinitionsFromCache() { + return + } + + if poller.state.Load() != nil { + return + } + + // Without definitions local evaluation is impossible, so fetch anyway. + poller.Logger.Debugf("[FEATURE FLAGS] No definitions cached or in memory, fetching from the API") + } + + poller.fetchFlagDefinitions(shouldFetch) +} + +// loadFlagDefinitionsFromCache applies cached definitions and reports whether any +// were loaded. +func (poller *FeatureFlagsPoller) loadFlagDefinitionsFromCache() bool { + data, err := poller.cacheProvider.GetFlagDefinitions(context.Background()) + if err != nil { + poller.Logger.Errorf("[FEATURE FLAGS] Cache provider GetFlagDefinitions failed: %s", err) + return false + } + if data == nil { + return false + } + if err := validateFlagDefinitions(*data); err != nil { + poller.Logger.Errorf("[FEATURE FLAGS] Cache provider returned unusable flag definitions: %s", err) + return false + } + + // The ETag is dropped: a later conditional request must not be answered with 304 + // against an ETag whose payload this instance no longer holds. + poller.applyFlagDefinitions(*data, "") + poller.Logger.Debugf("[FEATURE FLAGS] Loaded %d flag definitions from the external cache", len(data.Flags)) + return true +} + +// validateFlagDefinitions reports why cached definitions cannot be applied. +func validateFlagDefinitions(data FlagDefinitionCacheData) error { + if data.Flags == nil { + return errors.New("flags is missing") + } + + for _, flag := range data.Flags { + if flag.Filters.Multivariate == nil { + continue + } + for _, variant := range flag.Filters.Multivariate.Variants { + if variant.RolloutPercentage == nil { + return fmt.Errorf("flag %q has variant %q without a rollout percentage", flag.Key, variant.Key) + } + } + } + return nil +} + +// applyFlagDefinitions precomputes the evaluation lookups and atomically swaps in the +// new state. +func (poller *FeatureFlagsPoller) applyFlagDefinitions(data FlagDefinitionCacheData, etag string) { + newFlags := append(make([]FeatureFlag, 0, len(data.Flags)), data.Flags...) + preDecodePayloads(newFlags) + + groups := data.GroupTypeMapping + if groups == nil { + groups = map[string]string{} + } + + poller.state.Store(&flagsState{ + featureFlags: newFlags, + flagsByKey: buildFlagsByKey(newFlags), + cohorts: preParseCohortValues(data.Cohorts), + groups: groups, + flagsEtag: etag, + minimalFlagCalledEvents: data.MinimalFlagCalledEvents, + }) +} + +// publishFlagDefinitions stores definitions in the cache provider. +func (poller *FeatureFlagsPoller) publishFlagDefinitions(data FlagDefinitionCacheData) { + if err := poller.cacheProvider.OnFlagDefinitionsReceived(context.Background(), data); err != nil { + poller.Logger.Errorf("[FEATURE FLAGS] Cache provider OnFlagDefinitionsReceived failed: %s", err) + } +} + +// shutdownCacheProvider releases the cache provider. +func (poller *FeatureFlagsPoller) shutdownCacheProvider(ctx context.Context) { + if poller.cacheProvider == nil { + return + } + + if err := poller.cacheProvider.Shutdown(ctx); err != nil { + poller.Logger.Errorf("[FEATURE FLAGS] Cache provider Shutdown failed: %s", err) + } +} + +// fetchFlagDefinitions fetches the latest feature flag definitions from the PostHog API. +// When publish is true, a successful response is also stored in the cache provider. +func (poller *FeatureFlagsPoller) fetchFlagDefinitions(publish bool) { personalApiKey := poller.personalApiKey headers := http.Header{"Authorization": []string{"Bearer " + personalApiKey}} @@ -550,6 +668,16 @@ func (poller *FeatureFlagsPoller) fetchNewFeatureFlags() { } poller.state.Store(newState) } + // Republished so that a cache entry with a TTL does not expire while the + // definitions keep coming back unchanged. + if publish && currentState != nil { + poller.publishFlagDefinitions(FlagDefinitionCacheData{ + Flags: currentState.featureFlags, + GroupTypeMapping: currentState.groups, + Cohorts: currentState.cohorts, + MinimalFlagCalledEvents: currentState.minimalFlagCalledEvents, + }) + } return } @@ -581,16 +709,6 @@ func (poller *FeatureFlagsPoller) fetchNewFeatureFlags() { poller.Logger.Errorf("Unable to unmarshal response from api/feature_flag/local_evaluation: %s", err) return } - newFlags := append(make([]FeatureFlag, 0, len(featureFlagsResponse.Flags)), featureFlagsResponse.Flags...) - - // Pre-decode payloads once at load time (avoids per-evaluation json unquoting) - preDecodePayloads(newFlags) - - // Pre-build flagsByKey index for O(1) lookup during evaluation - flagsByKey := buildFlagsByKey(newFlags) - - // Store new ETag from response (clear if server stops sending) - newEtag := res.Header.Get("ETag") // Build new groups map groups := map[string]string{} @@ -598,18 +716,19 @@ func (poller *FeatureFlagsPoller) fetchNewFeatureFlags() { groups = *featureFlagsResponse.GroupTypeMapping } - // Pre-parse cohort values into typed structs (avoids per-evaluation reconstruction) - parsedCohorts := preParseCohortValues(featureFlagsResponse.Cohorts) + data := FlagDefinitionCacheData{ + Flags: featureFlagsResponse.Flags, + GroupTypeMapping: groups, + Cohorts: featureFlagsResponse.Cohorts, + MinimalFlagCalledEvents: featureFlagsResponse.MinimalFlagCalledEvents, + } - // Atomic swap of entire state - poller.state.Store(&flagsState{ - featureFlags: newFlags, - flagsByKey: flagsByKey, - cohorts: parsedCohorts, - groups: groups, - flagsEtag: newEtag, - minimalFlagCalledEvents: featureFlagsResponse.MinimalFlagCalledEvents, - }) + // Store new ETag from response (clear if server stops sending) + poller.applyFlagDefinitions(data, res.Header.Get("ETag")) + + if publish { + poller.publishFlagDefinitions(data) + } } // getMinimalFlagCalledEvents reports whether the local-evaluation payload @@ -2320,8 +2439,19 @@ func (poller *FeatureFlagsPoller) ForceReload() { poller.forceReload <- true } -func (poller *FeatureFlagsPoller) shutdownPoller() { +// shutdownPoller stops the polling loop and then releases the cache provider, so +// that no provider call is in flight when Shutdown runs. It gives up waiting when +// ctx is done, bounding cleanup by the deadline the caller passed to Close. +func (poller *FeatureFlagsPoller) shutdownPoller(ctx context.Context) { close(poller.shutdown) + + select { + case <-poller.shutdownDone: + case <-ctx.Done(): + poller.Logger.Warnf("[FEATURE FLAGS] Polling loop did not stop before the shutdown deadline: %s", ctx.Err()) + } + + poller.shutdownCacheProvider(ctx) } // getFeatureFlagVariants is a helper function to get the feature flag variants for diff --git a/flag_definition_cache.go b/flag_definition_cache.go new file mode 100644 index 0000000..0c2efa6 --- /dev/null +++ b/flag_definition_cache.go @@ -0,0 +1,34 @@ +package posthog + +import "context" + +// FlagDefinitionCacheData is the set of local evaluation data for a project. +type FlagDefinitionCacheData struct { + _ struct{} + + Flags []FeatureFlag `json:"flags"` + GroupTypeMapping map[string]string `json:"group_type_mapping"` + Cohorts map[string]PropertyGroup `json:"cohorts"` + MinimalFlagCalledEvents bool `json:"minimal_flag_called_events"` +} + +// FlagDefinitionCacheProvider shares feature flag definitions between SDK +// instances through an external cache such as Redis. +// +// EXPERIMENTAL: this interface may change in a minor version bump. +type FlagDefinitionCacheProvider interface { + // GetFlagDefinitions returns the cached flag definitions, or nil when nothing + // is cached. + GetFlagDefinitions(ctx context.Context) (*FlagDefinitionCacheData, error) + + // ShouldFetchFlagDefinitions reports whether this instance should fetch + // definitions from PostHog on this poll. + ShouldFetchFlagDefinitions(ctx context.Context) (bool, error) + + // OnFlagDefinitionsReceived stores definitions fetched from PostHog. + OnFlagDefinitionsReceived(ctx context.Context, data FlagDefinitionCacheData) error + + // Shutdown releases any resources held by the provider, such as a lock + // acquired by ShouldFetchFlagDefinitions. + Shutdown(ctx context.Context) error +} diff --git a/flag_definition_cache_test.go b/flag_definition_cache_test.go new file mode 100644 index 0000000..0a072af --- /dev/null +++ b/flag_definition_cache_test.go @@ -0,0 +1,746 @@ +package posthog + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +const cachedFlagDefinitions = `{ + "flags": [ + { + "key": "cached-flag", + "active": true, + "filters": { + "groups": [{"properties": [], "rollout_percentage": 100}], + "payloads": {"true": "{\"from\":\"cache\"}"} + } + }, + { + "key": "cached-multivariate", + "active": true, + "filters": { + "groups": [{"properties": [{"key": "id", "operator": "exact", "value": "1", "type": "cohort"}], "rollout_percentage": 100}], + "multivariate": {"variants": [{"key": "control", "name": "Control", "rollout_percentage": 100}]} + } + } + ], + "group_type_mapping": {"0": "company"}, + "cohorts": { + "1": { + "type": "OR", + "values": [{"key": "plan", "operator": "exact", "value": ["enterprise"], "type": "person"}] + } + }, + "minimal_flag_called_events": true +}` + +type fakeFlagDefinitionCache struct { + mu sync.Mutex + + shouldFetch bool + shouldFetchErr error + cached *FlagDefinitionCacheData + getErr error + publishErr error + shutdownErr error + onShouldFetch func() + onShutdown func(ctx context.Context) error + + shouldFetchCalls int + getCalls int + published []FlagDefinitionCacheData + shutdownCalls int +} + +func (c *fakeFlagDefinitionCache) ShouldFetchFlagDefinitions(context.Context) (bool, error) { + c.mu.Lock() + c.shouldFetchCalls++ + onShouldFetch := c.onShouldFetch + shouldFetch, err := c.shouldFetch, c.shouldFetchErr + c.mu.Unlock() + + if onShouldFetch != nil { + onShouldFetch() + } + return shouldFetch, err +} + +func (c *fakeFlagDefinitionCache) GetFlagDefinitions(context.Context) (*FlagDefinitionCacheData, error) { + c.mu.Lock() + defer c.mu.Unlock() + c.getCalls++ + return c.cached, c.getErr +} + +func (c *fakeFlagDefinitionCache) OnFlagDefinitionsReceived(_ context.Context, data FlagDefinitionCacheData) error { + c.mu.Lock() + defer c.mu.Unlock() + c.published = append(c.published, data) + return c.publishErr +} + +func (c *fakeFlagDefinitionCache) Shutdown(ctx context.Context) error { + c.mu.Lock() + c.shutdownCalls++ + onShutdown := c.onShutdown + err := c.shutdownErr + c.mu.Unlock() + + if onShutdown != nil { + return onShutdown(ctx) + } + return err +} + +func (c *fakeFlagDefinitionCache) calls() (shouldFetch, get, shutdown int, published []FlagDefinitionCacheData) { + c.mu.Lock() + defer c.mu.Unlock() + return c.shouldFetchCalls, c.getCalls, c.shutdownCalls, append([]FlagDefinitionCacheData(nil), c.published...) +} + +func definitionsServer(t *testing.T, handler func(w http.ResponseWriter, r *http.Request)) (*httptest.Server, func() int) { + t.Helper() + + var mu sync.Mutex + requests := 0 + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !strings.HasPrefix(r.URL.Path, "/flags/definitions") { + w.WriteHeader(http.StatusOK) + return + } + mu.Lock() + requests++ + mu.Unlock() + handler(w, r) + })) + t.Cleanup(server.Close) + + return server, func() int { + mu.Lock() + defer mu.Unlock() + return requests + } +} + +func serveDefinitions(body string) func(w http.ResponseWriter, r *http.Request) { + return func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("ETag", `"etag-1"`) + _, _ = w.Write([]byte(body)) + } +} + +// newCachingTestPoller builds a poller wired to a cache provider without starting the +// polling goroutine. +func newCachingTestPoller(t *testing.T, serverURL string, provider FlagDefinitionCacheProvider) *FeatureFlagsPoller { + t.Helper() + + poller := newTestPoller(t, serverURL) + poller.cacheProvider = provider + + poller.firstFeatureFlagRequestFinished = make(chan bool) + close(poller.firstFeatureFlagRequestFinished) + + poller.shutdownDone = make(chan bool) + close(poller.shutdownDone) + + return poller +} + +func TestFlagDefinitionCacheLeaderFetchesAndPublishes(t *testing.T) { + provider := &fakeFlagDefinitionCache{shouldFetch: true} + server, requests := definitionsServer(t, serveDefinitions(cachedFlagDefinitions)) + + poller := newCachingTestPoller(t, server.URL, provider) + poller.fetchNewFeatureFlags() + + require.Equal(t, 1, requests(), "the elected instance fetches from the API") + + shouldFetchCalls, getCalls, _, published := provider.calls() + require.Equal(t, 1, shouldFetchCalls) + require.Zero(t, getCalls, "the fetching instance has no reason to read the cache") + require.Len(t, published, 1, "fetched definitions are published for the other instances") + + require.Len(t, published[0].Flags, 2) + require.Equal(t, map[string]string{"0": "company"}, published[0].GroupTypeMapping) + require.Contains(t, published[0].Cohorts, "1") + require.True(t, published[0].MinimalFlagCalledEvents) + + state := poller.state.Load() + require.NotNil(t, state) + require.Len(t, state.featureFlags, 2) + require.Equal(t, `"etag-1"`, state.flagsEtag) +} + +func TestFlagDefinitionCacheFollowerReadsCacheInsteadOfAPI(t *testing.T) { + var cached FlagDefinitionCacheData + require.NoError(t, json.Unmarshal([]byte(cachedFlagDefinitions), &cached)) + + provider := &fakeFlagDefinitionCache{shouldFetch: false, cached: &cached} + server, requests := definitionsServer(t, serveDefinitions(cachedFlagDefinitions)) + + poller := newCachingTestPoller(t, server.URL, provider) + poller.fetchNewFeatureFlags() + + require.Zero(t, requests(), "a follower must not call the API") + + shouldFetchCalls, getCalls, _, published := provider.calls() + require.Equal(t, 1, shouldFetchCalls) + require.Equal(t, 1, getCalls) + require.Empty(t, published, "a follower must not publish definitions it did not fetch") + + state := poller.state.Load() + require.NotNil(t, state) + require.Len(t, state.featureFlags, 2) + require.Equal(t, map[string]string{"0": "company"}, state.groups) + require.True(t, state.minimalFlagCalledEvents, "the gate travels with the cached definitions") + require.Empty(t, state.flagsEtag, "cached definitions must not inherit an ETag") + + require.Contains(t, state.flagsByKey, "cached-flag") + require.Equal(t, `{"from":"cache"}`, state.flagsByKey["cached-flag"].Filters.DecodedPayloads["true"]) + require.Len(t, state.flagsByKey["cached-multivariate"].Filters.VariantLookupTable, 1) + require.Len(t, state.cohorts["1"].ParsedValues, 1) +} + +func TestFlagDefinitionCacheEvaluatesFlagsLoadedFromCache(t *testing.T) { + leader := &fakeFlagDefinitionCache{shouldFetch: true} + server, _ := definitionsServer(t, serveDefinitions(cachedFlagDefinitions)) + + leaderPoller := newCachingTestPoller(t, server.URL, leader) + leaderPoller.fetchNewFeatureFlags() + + _, _, _, published := leader.calls() + require.Len(t, published, 1) + + encoded, err := json.Marshal(published[0]) + require.NoError(t, err) + var roundTripped FlagDefinitionCacheData + require.NoError(t, json.Unmarshal(encoded, &roundTripped)) + + follower := &fakeFlagDefinitionCache{shouldFetch: false, cached: &roundTripped} + followerPoller := newCachingTestPoller(t, server.URL, follower) + followerPoller.fetchNewFeatureFlags() + + for name, poller := range map[string]*FeatureFlagsPoller{"leader": leaderPoller, "follower": followerPoller} { + value, isLocal, err := poller.GetFeatureFlag(FeatureFlagPayload{ + Key: "cached-flag", + DistinctId: "user-1", + OnlyEvaluateLocally: true, + }) + require.NoError(t, err, name) + require.True(t, isLocal, name) + require.Equal(t, true, value, name) + + payload, err := poller.GetFeatureFlagPayload(FeatureFlagPayload{ + Key: "cached-flag", + DistinctId: "user-1", + OnlyEvaluateLocally: true, + }) + require.NoError(t, err, name) + require.Equal(t, `{"from":"cache"}`, payload, name) + + variant, isLocal, err := poller.GetFeatureFlag(FeatureFlagPayload{ + Key: "cached-multivariate", + DistinctId: "user-1", + PersonProperties: Properties{"plan": "enterprise"}, + OnlyEvaluateLocally: true, + }) + require.NoError(t, err, name) + require.True(t, isLocal, name) + require.Equal(t, "control", variant, name) + } +} + +func TestFlagDefinitionCacheEmptyFlagsIsAHit(t *testing.T) { + provider := &fakeFlagDefinitionCache{ + shouldFetch: false, + cached: &FlagDefinitionCacheData{Flags: []FeatureFlag{}}, + } + server, requests := definitionsServer(t, serveDefinitions(cachedFlagDefinitions)) + + poller := newCachingTestPoller(t, server.URL, provider) + poller.fetchNewFeatureFlags() + + require.Zero(t, requests(), "empty definitions from the cache are a hit, not a miss") + + state := poller.state.Load() + require.NotNil(t, state) + require.Empty(t, state.featureFlags) +} + +func TestFlagDefinitionCacheMissingFlagsIsAMiss(t *testing.T) { + // What a provider that deserialized the JSON document `{}` hands back. + provider := &fakeFlagDefinitionCache{shouldFetch: false, cached: &FlagDefinitionCacheData{}} + server, requests := definitionsServer(t, serveDefinitions(cachedFlagDefinitions)) + + poller := newCachingTestPoller(t, server.URL, provider) + poller.fetchNewFeatureFlags() + + require.Equal(t, 1, requests(), "definitions without a flags key are unusable, so fetch instead") + + state := poller.state.Load() + require.NotNil(t, state) + require.Len(t, state.featureFlags, 2) +} + +func TestFlagDefinitionCacheUnusableDefinitionsKeepTheLoadedOnes(t *testing.T) { + malformed := FlagDefinitionCacheData{ + Flags: []FeatureFlag{{ + Key: "malformed-multivariate", + Active: true, + Filters: Filter{ + Multivariate: &Variants{Variants: []FlagVariant{{Key: "control"}}}, + }, + }}, + } + + provider := &fakeFlagDefinitionCache{shouldFetch: true} + server, requests := definitionsServer(t, serveDefinitions(cachedFlagDefinitions)) + + poller := newCachingTestPoller(t, server.URL, provider) + poller.fetchNewFeatureFlags() + require.Equal(t, 1, requests()) + + provider.mu.Lock() + provider.shouldFetch = false + provider.cached = &malformed + provider.mu.Unlock() + + require.NotPanics(t, poller.fetchNewFeatureFlags) + + require.Equal(t, 1, requests(), "warm definitions are kept, so there is nothing to recover") + state := poller.state.Load() + require.NotNil(t, state) + require.Len(t, state.featureFlags, 2) + require.Contains(t, state.flagsByKey, "cached-flag") +} + +func TestFlagDefinitionCacheUnusableDefinitionsFetchWithoutWarmState(t *testing.T) { + provider := &fakeFlagDefinitionCache{ + shouldFetch: false, + cached: &FlagDefinitionCacheData{ + Flags: []FeatureFlag{{ + Key: "malformed-multivariate", + Active: true, + Filters: Filter{Multivariate: &Variants{Variants: []FlagVariant{{Key: "control"}}}}, + }}, + }, + } + server, requests := definitionsServer(t, serveDefinitions(cachedFlagDefinitions)) + + poller := newCachingTestPoller(t, server.URL, provider) + require.NotPanics(t, poller.fetchNewFeatureFlags) + + require.Equal(t, 1, requests()) + state := poller.state.Load() + require.NotNil(t, state) + require.Len(t, state.featureFlags, 2) +} + +func TestFlagDefinitionCacheMissWithoutDefinitionsFetchesAnyway(t *testing.T) { + provider := &fakeFlagDefinitionCache{shouldFetch: false, cached: nil} + server, requests := definitionsServer(t, serveDefinitions(cachedFlagDefinitions)) + + poller := newCachingTestPoller(t, server.URL, provider) + poller.fetchNewFeatureFlags() + + require.Equal(t, 1, requests()) + + _, getCalls, _, published := provider.calls() + require.Equal(t, 1, getCalls) + require.Empty(t, published, "an instance that was told not to fetch must not publish") + + state := poller.state.Load() + require.NotNil(t, state) + require.Len(t, state.featureFlags, 2) +} + +func TestFlagDefinitionCacheMissKeepsDefinitionsAlreadyLoaded(t *testing.T) { + provider := &fakeFlagDefinitionCache{shouldFetch: true} + server, requests := definitionsServer(t, serveDefinitions(cachedFlagDefinitions)) + + poller := newCachingTestPoller(t, server.URL, provider) + poller.fetchNewFeatureFlags() + require.Equal(t, 1, requests()) + + provider.mu.Lock() + provider.shouldFetch = false + provider.cached = nil + provider.mu.Unlock() + + poller.fetchNewFeatureFlags() + + require.Equal(t, 1, requests(), "stale definitions are preferred over ignoring the cache provider") + state := poller.state.Load() + require.NotNil(t, state) + require.Len(t, state.featureFlags, 2) +} + +func TestFlagDefinitionCacheShouldFetchErrorFallsBackToAPI(t *testing.T) { + provider := &fakeFlagDefinitionCache{shouldFetch: false, shouldFetchErr: errors.New("redis down")} + server, requests := definitionsServer(t, serveDefinitions(cachedFlagDefinitions)) + + poller := newCachingTestPoller(t, server.URL, provider) + poller.fetchNewFeatureFlags() + + require.Equal(t, 1, requests()) + + _, getCalls, _, published := provider.calls() + require.Zero(t, getCalls) + require.Len(t, published, 1, "the fallback fetch is treated as a normal fetch") + + state := poller.state.Load() + require.NotNil(t, state) + require.Len(t, state.featureFlags, 2) +} + +func TestFlagDefinitionCacheGetErrorIsTreatedAsAMiss(t *testing.T) { + provider := &fakeFlagDefinitionCache{shouldFetch: false, getErr: errors.New("redis down")} + server, requests := definitionsServer(t, serveDefinitions(cachedFlagDefinitions)) + + poller := newCachingTestPoller(t, server.URL, provider) + + poller.fetchNewFeatureFlags() + require.Equal(t, 1, requests()) + require.NotNil(t, poller.state.Load()) + + poller.fetchNewFeatureFlags() + require.Equal(t, 1, requests()) + require.Len(t, poller.state.Load().featureFlags, 2) +} + +func TestFlagDefinitionCachePublishErrorKeepsDefinitionsInMemory(t *testing.T) { + provider := &fakeFlagDefinitionCache{shouldFetch: true, publishErr: errors.New("redis down")} + server, requests := definitionsServer(t, serveDefinitions(cachedFlagDefinitions)) + + poller := newCachingTestPoller(t, server.URL, provider) + poller.fetchNewFeatureFlags() + + require.Equal(t, 1, requests()) + state := poller.state.Load() + require.NotNil(t, state, "a cache that cannot store definitions does not stop this instance") + require.Len(t, state.featureFlags, 2) +} + +func TestFlagDefinitionCacheNotModifiedRepublishesDefinitions(t *testing.T) { + var requestCount int + server, requests := definitionsServer(t, func(w http.ResponseWriter, r *http.Request) { + requestCount++ + w.Header().Set("ETag", `"etag-1"`) + if requestCount == 1 { + _, _ = w.Write([]byte(cachedFlagDefinitions)) + return + } + require.Equal(t, `"etag-1"`, r.Header.Get("If-None-Match")) + w.WriteHeader(http.StatusNotModified) + }) + + provider := &fakeFlagDefinitionCache{shouldFetch: true} + poller := newCachingTestPoller(t, server.URL, provider) + + poller.fetchNewFeatureFlags() + poller.fetchNewFeatureFlags() + + require.Equal(t, 2, requests()) + + _, _, _, published := provider.calls() + require.Len(t, published, 2) + require.Len(t, published[1].Flags, 2, "the 304 republishes the definitions in memory") + require.Contains(t, published[1].Cohorts, "1") + require.True(t, published[1].MinimalFlagCalledEvents, "the 304 republishes the gate too") +} + +func TestFlagDefinitionCacheNotModifiedDoesNotPublishForFollowers(t *testing.T) { + server, _ := definitionsServer(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("ETag", `"etag-1"`) + if r.Header.Get("If-None-Match") == `"etag-1"` { + w.WriteHeader(http.StatusNotModified) + return + } + _, _ = w.Write([]byte(cachedFlagDefinitions)) + }) + + provider := &fakeFlagDefinitionCache{shouldFetch: false} + poller := newCachingTestPoller(t, server.URL, provider) + poller.fetchNewFeatureFlags() + + _, _, _, published := provider.calls() + require.Empty(t, published) +} + +func TestFlagDefinitionCacheQuotaLimitedDoesNotPublish(t *testing.T) { + provider := &fakeFlagDefinitionCache{shouldFetch: true} + server, requests := definitionsServer(t, func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusPaymentRequired) + }) + + poller := newCachingTestPoller(t, server.URL, provider) + poller.fetchNewFeatureFlags() + + require.Equal(t, 1, requests()) + _, _, _, published := provider.calls() + require.Empty(t, published) +} + +func TestFlagDefinitionCacheProviderNotCalledWithoutConfiguration(t *testing.T) { + server, requests := definitionsServer(t, serveDefinitions(cachedFlagDefinitions)) + + poller := newTestPoller(t, server.URL) + poller.fetchNewFeatureFlags() + + require.Equal(t, 1, requests()) + require.NotNil(t, poller.state.Load()) +} + +func TestFlagDefinitionCacheShutdownReleasesProvider(t *testing.T) { + provider := &fakeFlagDefinitionCache{shouldFetch: true} + server, _ := definitionsServer(t, serveDefinitions(cachedFlagDefinitions)) + + poller, err := newFeatureFlagsPoller( + "test-api-key", + "test-personal-key", + newDefaultLogger(false), + server.URL, + http.Client{}, + time.Hour, + nil, + 10*time.Second, + nil, + false, + provider, + ) + require.NoError(t, err) + + <-poller.firstFeatureFlagRequestFinished + poller.shutdownPoller(context.Background()) + + _, _, shutdownCalls, _ := provider.calls() + require.Equal(t, 1, shutdownCalls, "shutdownPoller waits for the provider to be released") + + select { + case <-poller.shutdownDone: + default: + t.Fatal("Shutdown ran while the polling loop was still running") + } +} + +func TestFlagDefinitionCacheShutdownErrorDoesNotBlockShutdown(t *testing.T) { + provider := &fakeFlagDefinitionCache{shouldFetch: true, shutdownErr: errors.New("redis down")} + server, _ := definitionsServer(t, serveDefinitions(cachedFlagDefinitions)) + + poller := newCachingTestPoller(t, server.URL, provider) + poller.shutdown = make(chan bool) + + done := make(chan struct{}) + go func() { + poller.shutdownPoller(context.Background()) + close(done) + }() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("shutdownPoller did not return after a failing provider shutdown") + } + + _, _, shutdownCalls, _ := provider.calls() + require.Equal(t, 1, shutdownCalls) +} + +func TestFlagDefinitionCacheShutdownUsesTheCallerDeadline(t *testing.T) { + var deadline time.Time + var hasDeadline bool + + provider := &fakeFlagDefinitionCache{ + shouldFetch: true, + onShutdown: func(ctx context.Context) error { + deadline, hasDeadline = ctx.Deadline() + return nil + }, + } + server, _ := definitionsServer(t, serveDefinitions(cachedFlagDefinitions)) + + poller := newCachingTestPoller(t, server.URL, provider) + poller.shutdown = make(chan bool) + + callerDeadline := time.Now().Add(20 * time.Millisecond) + ctx, cancel := context.WithDeadline(context.Background(), callerDeadline) + defer cancel() + poller.shutdownPoller(ctx) + + require.True(t, hasDeadline, "Shutdown was handed a context with no deadline") + require.Equal(t, callerDeadline, deadline, "the provider must not get its own, longer deadline") +} + +func TestFlagDefinitionCacheShutdownWaitsForAnInFlightProviderCall(t *testing.T) { + fetching := make(chan struct{}) + release := make(chan struct{}) + shutdownStarted := make(chan struct{}) + + var inFlight atomic.Int32 + var inFlightAtShutdown int32 + + provider := &fakeFlagDefinitionCache{ + shouldFetch: true, + onShouldFetch: func() { + inFlight.Add(1) + defer inFlight.Add(-1) + close(fetching) + <-release + }, + onShutdown: func(context.Context) error { + inFlightAtShutdown = inFlight.Load() + close(shutdownStarted) + return nil + }, + } + server, _ := definitionsServer(t, serveDefinitions(cachedFlagDefinitions)) + + poller, err := newFeatureFlagsPoller( + "test-api-key", + "test-personal-key", + newDefaultLogger(false), + server.URL, + http.Client{}, + time.Hour, + nil, + 10*time.Second, + nil, + false, + provider, + ) + require.NoError(t, err) + + <-fetching + + done := make(chan struct{}) + go func() { + poller.shutdownPoller(context.Background()) + close(done) + }() + + select { + case <-shutdownStarted: + t.Fatal("Shutdown ran while a provider call was still in flight") + case <-time.After(100 * time.Millisecond): + } + + close(release) + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("shutdownPoller never returned") + } + + require.Zero(t, inFlightAtShutdown, "Shutdown must not overlap another provider call") +} + +func TestFlagDefinitionCacheCloseDeadlineReachesTheProvider(t *testing.T) { + entered := make(chan struct{}) + provider := &fakeFlagDefinitionCache{ + shouldFetch: true, + onShutdown: func(ctx context.Context) error { + close(entered) + <-ctx.Done() + return ctx.Err() + }, + } + server, _ := definitionsServer(t, serveDefinitions(cachedFlagDefinitions)) + + client, err := NewWithConfig("phc_test", Config{ + Endpoint: server.URL, + SecretKey: "phs_test", + Interval: time.Millisecond, + DefaultFeatureFlagsPollingInterval: time.Hour, + ShutdownTimeout: 20 * time.Millisecond, + FlagDefinitionCacheProvider: provider, + }) + require.NoError(t, err) + + closed := make(chan error, 1) + go func() { closed <- client.Close() }() + + select { + case err := <-closed: + require.ErrorContains(t, err, "shutdown timeout") + case <-time.After(5 * time.Second): + t.Fatal("Close hung on a provider that waits for its context") + } + + select { + case <-entered: + default: + t.Fatal("the provider was never asked to shut down") + } +} + +func TestFlagDefinitionCacheShutdownStopsWaitingOnADeadDeadline(t *testing.T) { + provider := &fakeFlagDefinitionCache{shouldFetch: true} + server, _ := definitionsServer(t, serveDefinitions(cachedFlagDefinitions)) + + poller := newCachingTestPoller(t, server.URL, provider) + poller.shutdown = make(chan bool) + // A polling loop that never returns. + poller.shutdownDone = make(chan bool) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + done := make(chan struct{}) + go func() { + poller.shutdownPoller(ctx) + close(done) + }() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("shutdownPoller kept waiting for the polling loop past the deadline") + } + + _, _, shutdownCalls, _ := provider.calls() + require.Equal(t, 1, shutdownCalls) +} + +func TestFlagDefinitionCacheProviderThroughClient(t *testing.T) { + var cached FlagDefinitionCacheData + require.NoError(t, json.Unmarshal([]byte(cachedFlagDefinitions), &cached)) + + provider := &fakeFlagDefinitionCache{shouldFetch: false, cached: &cached} + server, requests := definitionsServer(t, serveDefinitions(cachedFlagDefinitions)) + + client, err := NewWithConfig("phc_test", Config{ + Endpoint: server.URL, + SecretKey: "phs_test", + Interval: time.Millisecond, + DefaultFeatureFlagsPollingInterval: time.Hour, + FlagDefinitionCacheProvider: provider, + }) + require.NoError(t, err) + + value, err := client.GetFeatureFlag(FeatureFlagPayload{ + Key: "cached-flag", + DistinctId: "user-1", + OnlyEvaluateLocally: true, + }) + require.NoError(t, err) + require.Equal(t, true, value) + require.Zero(t, requests(), "definitions came from the cache, not the API") + + require.NoError(t, client.Close()) + + _, getCalls, shutdownCalls, _ := provider.calls() + require.GreaterOrEqual(t, getCalls, 1) + require.Equal(t, 1, shutdownCalls, "closing the client releases the provider") +} diff --git a/posthog.go b/posthog.go index 86aa231..7133171 100644 --- a/posthog.go +++ b/posthog.go @@ -298,6 +298,7 @@ func NewWithConfig(apiKey string, config Config) (cli Client, err error) { c.FeatureFlagRequestTimeout, c.decider, c.Config.GetDisableGeoIP(), + c.FlagDefinitionCacheProvider, ) if err != nil { return nil, err @@ -1724,7 +1725,7 @@ func (c *client) loop() { defer close(c.batches) // prevent any pending receives from blocking defer close(c.shutdown) if c.featureFlagsPoller != nil { - defer c.featureFlagsPoller.shutdownPoller() + defer c.featureFlagsPoller.shutdownPoller(c.ctx) } var batchData []json.RawMessage