From 1490adaf656c8745d47bbef71602dfd0dd901be6 Mon Sep 17 00:00:00 2001 From: Andrey Vasenin Date: Wed, 2 Sep 2026 18:48:51 +0200 Subject: [PATCH 1/9] Add cache support --- api/public-api.txt | 19 ++ config.go | 6 + examples/README.md | 1 + examples/flag_definition_cache.go | 209 ++++++++++++ examples/main.go | 21 +- featureflags.go | 156 +++++++-- flag_definition_cache.go | 31 ++ flag_definition_cache_test.go | 523 ++++++++++++++++++++++++++++++ posthog.go | 1 + 9 files changed, 932 insertions(+), 35 deletions(-) create mode 100644 examples/flag_definition_cache.go create mode 100644 flag_definition_cache.go create mode 100644 flag_definition_cache_test.go diff --git a/api/public-api.txt b/api/public-api.txt index e18fd5d..4ba3a1b 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,23 @@ 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"` +} + +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..4134572 100644 --- a/config.go +++ b/config.go @@ -126,6 +126,12 @@ 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. + // + // 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..6fc0498 --- /dev/null +++ b/examples/flag_definition_cache.go @@ -0,0 +1,209 @@ +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 +) + +// 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 +// should 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 + } + + switch { + case holder == c.instanceID: + // Already the leader: extend the lock. + case holder == "": + case time.Now().Before(expiry): + return false, nil + } + + return true, c.writeLock() +} + +// GetFlagDefinitions returns the published definitions, or nil when there are none. +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()) +} + +// Shutdown releases the lock if this instance holds it. +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 two 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("Two instances sharing the cache in %s\n\n", dir) + + clients := make([]posthog.Client, 0, 2) + caches := make([]*FileFlagCache, 0, 2) + + for i := 0; i < 4; 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 a flag on both instances, 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..1440ead 100644 --- a/featureflags.go +++ b/featureflags.go @@ -64,6 +64,8 @@ func getOrCompileRegex(pattern string) (*regexp.Regexp, error) { return r, nil } +const flagDefinitionCacheShutdownTimeout = 30 * time.Second + // flagsState holds the feature flag data that is atomically swapped during updates. // This provides lock-free reads for the common path (flag evaluation). type flagsState struct { @@ -95,12 +97,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 +464,7 @@ func newFeatureFlagsPoller( flagTimeout time.Duration, decider decider, disableGeoIP bool, + cacheProvider FlagDefinitionCacheProvider, ) (*FeatureFlagsPoller, error) { localEvaluationEndpoint := "/flags/definitions" localEvalURL, err := url.Parse(endpoint + localEvaluationEndpoint) @@ -486,6 +490,7 @@ func newFeatureFlagsPoller( flagTimeout: flagTimeout, decider: decider, disableGeoIP: disableGeoIP, + cacheProvider: cacheProvider, } go poller.run() @@ -512,10 +517,101 @@ 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 + } + + // 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, "", false) + poller.Logger.Debugf("[FEATURE FLAGS] Loaded %d flag definitions from the external cache", len(data.Flags)) + return true +} + +// applyFlagDefinitions precomputes the evaluation lookups and atomically swaps in the +// new state. +func (poller *FeatureFlagsPoller) applyFlagDefinitions(data FlagDefinitionCacheData, etag string, minimalFlagCalledEvents bool) { + 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: 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() { + if poller.cacheProvider == nil { + return + } + + ctx, cancel := context.WithTimeout(context.Background(), flagDefinitionCacheShutdownTimeout) + defer cancel() + + 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 +646,15 @@ 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, + }) + } return } @@ -581,16 +686,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 +693,18 @@ 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, + } - // 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"), featureFlagsResponse.MinimalFlagCalledEvents) + + if publish { + poller.publishFlagDefinitions(data) + } } // getMinimalFlagCalledEvents reports whether the local-evaluation payload @@ -2322,6 +2417,7 @@ func (poller *FeatureFlagsPoller) ForceReload() { func (poller *FeatureFlagsPoller) shutdownPoller() { close(poller.shutdown) + poller.shutdownCacheProvider() } // 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..2edb8d1 --- /dev/null +++ b/flag_definition_cache.go @@ -0,0 +1,31 @@ +package posthog + +import "context" + +// FlagDefinitionCacheData is the set of local evaluation data for a project. +type FlagDefinitionCacheData struct { + Flags []FeatureFlag `json:"flags"` + GroupTypeMapping map[string]string `json:"group_type_mapping"` + Cohorts map[string]PropertyGroup `json:"cohorts"` +} + +// 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..acc8749 --- /dev/null +++ b/flag_definition_cache_test.go @@ -0,0 +1,523 @@ +package posthog + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync" + "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 + 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() + defer c.mu.Unlock() + c.shouldFetchCalls++ + return c.shouldFetch, c.shouldFetchErr +} + +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) + + 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") + + 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.False(t, state.minimalFlagCalledEvents, "the gate is not shared, so cached definitions fall back to full events") + 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 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") +} + +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() + + _, _, shutdownCalls, _ := provider.calls() + require.Equal(t, 1, shutdownCalls, "shutdownPoller waits for the provider to be released") +} + +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() + 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 TestFlagDefinitionCacheShutdownContextCarriesADeadline(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) + poller.shutdownPoller() + + require.True(t, hasDeadline, "Shutdown was handed a context with no deadline") + require.WithinDuration(t, time.Now().Add(flagDefinitionCacheShutdownTimeout), deadline, time.Minute) +} + +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..cb8072a 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 From 28c9163690012ebeabcd7b953c57116abdb0a869 Mon Sep 17 00:00:00 2001 From: Andrey Vasenin Date: Sat, 5 Sep 2026 15:08:37 +0200 Subject: [PATCH 2/9] fix(flags): address review of the flag definition cache provider Share the server-controlled minimal_flag_called_events gate through the cached payload so that followers emit the same $feature_flag_called events as the instance that fetched the definitions. Validate provider data at the cache boundary: definitions without a flags key, and multivariate variants without a rollout percentage, are logged and treated as a cache miss instead of erasing the definitions already loaded or panicking during preprocessing. Order and bound provider cleanup. shutdownPoller now waits for the polling loop to return before calling Shutdown, and uses the client context rather than its own 30 second budget, so a short CloseWithContext deadline is honoured. Force keyed literals on FlagDefinitionCacheData, add the changeset, and fix the instance count in the example. Co-Authored-By: Claude --- .changeset/flag-definition-cache-provider.md | 5 + api/public-api.txt | 1 + examples/flag_definition_cache.go | 2 +- featureflags.go | 71 ++++++++--- flag_definition_cache.go | 6 + flag_definition_cache_test.go | 124 ++++++++++++++++++- posthog.go | 2 +- 7 files changed, 185 insertions(+), 26 deletions(-) create mode 100644 .changeset/flag-definition-cache-provider.md diff --git a/.changeset/flag-definition-cache-provider.md b/.changeset/flag-definition-cache-provider.md new file mode 100644 index 0000000..b68c334 --- /dev/null +++ b/.changeset/flag-definition-cache-provider.md @@ -0,0 +1,5 @@ +--- +"posthog-go": minor +--- + +Add `Config.FlagDefinitionCacheProvider`, an experimental interface for sharing local evaluation flag definitions between SDK instances through an external cache such as Redis. One instance is elected to poll the PostHog API and publishes the definitions it fetches; the others load them from the cache instead of calling the API. The cached payload carries the flags, group type mapping, cohorts, and the server-controlled `minimal_flag_called_events` gate, so followers emit the same `$feature_flag_called` events as the instance that fetched. Cached definitions that cannot be used are ignored in favour of the definitions already loaded, and closing the client stops the polling loop before releasing the provider, bounded by the deadline passed to `CloseWithContext`. diff --git a/api/public-api.txt b/api/public-api.txt index 4ba3a1b..ea7c580 100644 --- a/api/public-api.txt +++ b/api/public-api.txt @@ -630,6 +630,7 @@ 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 { diff --git a/examples/flag_definition_cache.go b/examples/flag_definition_cache.go index 6fc0498..b5c29c9 100644 --- a/examples/flag_definition_cache.go +++ b/examples/flag_definition_cache.go @@ -161,7 +161,7 @@ func TestFlagDefinitionCache(projectAPIKey, secretKey, endpoint string) { clients := make([]posthog.Client, 0, 2) caches := make([]*FileFlagCache, 0, 2) - for i := 0; i < 4; i++ { + for i := 0; i < 2; i++ { cache, err := NewFileFlagCache(dir, "my-service-production") if err != nil { fmt.Printf("❌ Could not create the cache provider: %v\n", err) diff --git a/featureflags.go b/featureflags.go index 1440ead..37465f0 100644 --- a/featureflags.go +++ b/featureflags.go @@ -64,8 +64,6 @@ func getOrCompileRegex(pattern string) (*regexp.Regexp, error) { return r, nil } -const flagDefinitionCacheShutdownTimeout = 30 * time.Second - // flagsState holds the feature flag data that is atomically swapped during updates. // This provides lock-free reads for the common path (flag evaluation). type flagsState struct { @@ -87,6 +85,8 @@ type FeatureFlagsPoller struct { firstFeatureFlagRequestFinished chan bool shutdown chan bool forceReload chan bool + // pollLoopDone is closed when the polling goroutine has returned. + pollLoopDone chan struct{} // state holds all flag-related data using atomic pointer for lock-free reads state atomic.Pointer[flagsState] @@ -480,6 +480,7 @@ func newFeatureFlagsPoller( firstFeatureFlagRequestFinished: make(chan bool), shutdown: make(chan bool), forceReload: make(chan bool), + pollLoopDone: make(chan struct{}), personalApiKey: personalApiKey, projectApiKey: projectApiKey, localEvalUrl: localEvalURL, @@ -498,6 +499,8 @@ func newFeatureFlagsPoller( } func (poller *FeatureFlagsPoller) run() { + defer close(poller.pollLoopDone) + poller.fetchNewFeatureFlags() close(poller.firstFeatureFlagRequestFinished) @@ -559,17 +562,40 @@ func (poller *FeatureFlagsPoller) loadFlagDefinitionsFromCache() bool { 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, "", false) + 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, minimalFlagCalledEvents bool) { +func (poller *FeatureFlagsPoller) applyFlagDefinitions(data FlagDefinitionCacheData, etag string) { newFlags := append(make([]FeatureFlag, 0, len(data.Flags)), data.Flags...) preDecodePayloads(newFlags) @@ -584,7 +610,7 @@ func (poller *FeatureFlagsPoller) applyFlagDefinitions(data FlagDefinitionCacheD cohorts: preParseCohortValues(data.Cohorts), groups: groups, flagsEtag: etag, - minimalFlagCalledEvents: minimalFlagCalledEvents, + minimalFlagCalledEvents: data.MinimalFlagCalledEvents, }) } @@ -596,14 +622,11 @@ func (poller *FeatureFlagsPoller) publishFlagDefinitions(data FlagDefinitionCach } // shutdownCacheProvider releases the cache provider. -func (poller *FeatureFlagsPoller) shutdownCacheProvider() { +func (poller *FeatureFlagsPoller) shutdownCacheProvider(ctx context.Context) { if poller.cacheProvider == nil { return } - ctx, cancel := context.WithTimeout(context.Background(), flagDefinitionCacheShutdownTimeout) - defer cancel() - if err := poller.cacheProvider.Shutdown(ctx); err != nil { poller.Logger.Errorf("[FEATURE FLAGS] Cache provider Shutdown failed: %s", err) } @@ -650,9 +673,10 @@ func (poller *FeatureFlagsPoller) fetchFlagDefinitions(publish bool) { // definitions keep coming back unchanged. if publish && currentState != nil { poller.publishFlagDefinitions(FlagDefinitionCacheData{ - Flags: currentState.featureFlags, - GroupTypeMapping: currentState.groups, - Cohorts: currentState.cohorts, + Flags: currentState.featureFlags, + GroupTypeMapping: currentState.groups, + Cohorts: currentState.cohorts, + MinimalFlagCalledEvents: currentState.minimalFlagCalledEvents, }) } return @@ -694,13 +718,14 @@ func (poller *FeatureFlagsPoller) fetchFlagDefinitions(publish bool) { } data := FlagDefinitionCacheData{ - Flags: featureFlagsResponse.Flags, - GroupTypeMapping: groups, - Cohorts: featureFlagsResponse.Cohorts, + Flags: featureFlagsResponse.Flags, + GroupTypeMapping: groups, + Cohorts: featureFlagsResponse.Cohorts, + MinimalFlagCalledEvents: featureFlagsResponse.MinimalFlagCalledEvents, } // Store new ETag from response (clear if server stops sending) - poller.applyFlagDefinitions(data, res.Header.Get("ETag"), featureFlagsResponse.MinimalFlagCalledEvents) + poller.applyFlagDefinitions(data, res.Header.Get("ETag")) if publish { poller.publishFlagDefinitions(data) @@ -2415,9 +2440,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) - poller.shutdownCacheProvider() + + select { + case <-poller.pollLoopDone: + 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 index 2edb8d1..268d931 100644 --- a/flag_definition_cache.go +++ b/flag_definition_cache.go @@ -4,9 +4,15 @@ import "context" // FlagDefinitionCacheData is the set of local evaluation data for a project. type FlagDefinitionCacheData struct { + // _ forces keyed literals so that later fields do not break implementations. + _ struct{} + Flags []FeatureFlag `json:"flags"` GroupTypeMapping map[string]string `json:"group_type_mapping"` Cohorts map[string]PropertyGroup `json:"cohorts"` + // MinimalFlagCalledEvents is the server-controlled gate for minimal + // $feature_flag_called events. Absent means false. + MinimalFlagCalledEvents bool `json:"minimal_flag_called_events"` } // FlagDefinitionCacheProvider shares feature flag definitions between SDK diff --git a/flag_definition_cache_test.go b/flag_definition_cache_test.go index acc8749..f3115bc 100644 --- a/flag_definition_cache_test.go +++ b/flag_definition_cache_test.go @@ -143,6 +143,9 @@ func newCachingTestPoller(t *testing.T, serverURL string, provider FlagDefinitio poller.firstFeatureFlagRequestFinished = make(chan bool) close(poller.firstFeatureFlagRequestFinished) + poller.pollLoopDone = make(chan struct{}) + close(poller.pollLoopDone) + return poller } @@ -163,6 +166,7 @@ func TestFlagDefinitionCacheLeaderFetchesAndPublishes(t *testing.T) { 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) @@ -191,7 +195,7 @@ func TestFlagDefinitionCacheFollowerReadsCacheInsteadOfAPI(t *testing.T) { require.NotNil(t, state) require.Len(t, state.featureFlags, 2) require.Equal(t, map[string]string{"0": "company"}, state.groups) - require.False(t, state.minimalFlagCalledEvents, "the gate is not shared, so cached definitions fall back to full events") + 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") @@ -266,6 +270,75 @@ func TestFlagDefinitionCacheEmptyFlagsIsAHit(t *testing.T) { 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)) @@ -376,6 +449,7 @@ func TestFlagDefinitionCacheNotModifiedRepublishesDefinitions(t *testing.T) { 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) { @@ -440,10 +514,16 @@ func TestFlagDefinitionCacheShutdownReleasesProvider(t *testing.T) { require.NoError(t, err) <-poller.firstFeatureFlagRequestFinished - poller.shutdownPoller() + poller.shutdownPoller(context.Background()) _, _, shutdownCalls, _ := provider.calls() require.Equal(t, 1, shutdownCalls, "shutdownPoller waits for the provider to be released") + + select { + case <-poller.pollLoopDone: + default: + t.Fatal("Shutdown ran while the polling loop was still running") + } } func TestFlagDefinitionCacheShutdownErrorDoesNotBlockShutdown(t *testing.T) { @@ -455,7 +535,7 @@ func TestFlagDefinitionCacheShutdownErrorDoesNotBlockShutdown(t *testing.T) { done := make(chan struct{}) go func() { - poller.shutdownPoller() + poller.shutdownPoller(context.Background()) close(done) }() @@ -469,7 +549,7 @@ func TestFlagDefinitionCacheShutdownErrorDoesNotBlockShutdown(t *testing.T) { require.Equal(t, 1, shutdownCalls) } -func TestFlagDefinitionCacheShutdownContextCarriesADeadline(t *testing.T) { +func TestFlagDefinitionCacheShutdownUsesTheCallerDeadline(t *testing.T) { var deadline time.Time var hasDeadline bool @@ -484,10 +564,42 @@ func TestFlagDefinitionCacheShutdownContextCarriesADeadline(t *testing.T) { poller := newCachingTestPoller(t, server.URL, provider) poller.shutdown = make(chan bool) - poller.shutdownPoller() + + 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.WithinDuration(t, time.Now().Add(flagDefinitionCacheShutdownTimeout), deadline, time.Minute) + require.Equal(t, callerDeadline, deadline, "the provider must not get its own, longer deadline") +} + +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.pollLoopDone = make(chan struct{}) + + 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) { diff --git a/posthog.go b/posthog.go index cb8072a..7133171 100644 --- a/posthog.go +++ b/posthog.go @@ -1725,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 From e6f1b02647cc8275cab3337b08ff9c1466cd0011 Mon Sep 17 00:00:00 2001 From: Andrey Vasenin Date: Sat, 5 Sep 2026 15:17:20 +0200 Subject: [PATCH 3/9] style(flags): keep FlagDefinitionCacheData fields gofmt aligned Co-Authored-By: Claude --- api/public-api.txt | 8 ++++---- flag_definition_cache.go | 11 ++++------- 2 files changed, 8 insertions(+), 11 deletions(-) diff --git a/api/public-api.txt b/api/public-api.txt index ea7c580..c98bbcc 100644 --- a/api/public-api.txt +++ b/api/public-api.txt @@ -627,10 +627,10 @@ type Filter struct { } 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"` + 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 { diff --git a/flag_definition_cache.go b/flag_definition_cache.go index 268d931..0c2efa6 100644 --- a/flag_definition_cache.go +++ b/flag_definition_cache.go @@ -4,15 +4,12 @@ import "context" // FlagDefinitionCacheData is the set of local evaluation data for a project. type FlagDefinitionCacheData struct { - // _ forces keyed literals so that later fields do not break implementations. _ struct{} - Flags []FeatureFlag `json:"flags"` - GroupTypeMapping map[string]string `json:"group_type_mapping"` - Cohorts map[string]PropertyGroup `json:"cohorts"` - // MinimalFlagCalledEvents is the server-controlled gate for minimal - // $feature_flag_called events. Absent means false. - MinimalFlagCalledEvents bool `json:"minimal_flag_called_events"` + 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 From 1b3348ff0739ad98a9a68b85a4acb89b24570136 Mon Sep 17 00:00:00 2001 From: Andrey Vasenin Date: Sat, 5 Sep 2026 15:26:44 +0200 Subject: [PATCH 4/9] docs(flags): trim the changeset and link the distributed environments guide Co-Authored-By: Claude --- .changeset/flag-definition-cache-provider.md | 2 +- config.go | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/.changeset/flag-definition-cache-provider.md b/.changeset/flag-definition-cache-provider.md index b68c334..71880b3 100644 --- a/.changeset/flag-definition-cache-provider.md +++ b/.changeset/flag-definition-cache-provider.md @@ -2,4 +2,4 @@ "posthog-go": minor --- -Add `Config.FlagDefinitionCacheProvider`, an experimental interface for sharing local evaluation flag definitions between SDK instances through an external cache such as Redis. One instance is elected to poll the PostHog API and publishes the definitions it fetches; the others load them from the cache instead of calling the API. The cached payload carries the flags, group type mapping, cohorts, and the server-controlled `minimal_flag_called_events` gate, so followers emit the same `$feature_flag_called` events as the instance that fetched. Cached definitions that cannot be used are ignored in favour of the definitions already loaded, and closing the client stops the polling loop before releasing the provider, bounded by the deadline passed to `CloseWithContext`. +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/config.go b/config.go index 4134572..1651b7b 100644 --- a/config.go +++ b/config.go @@ -128,6 +128,8 @@ type Config struct { // 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 From 23c934b61f0bbfe69458a785f600a10063e5aefa Mon Sep 17 00:00:00 2001 From: Andrey Vasenin Date: Sat, 5 Sep 2026 15:36:11 +0200 Subject: [PATCH 5/9] test(flags): cover provider shutdown ordering and the close deadline Co-Authored-By: Claude --- flag_definition_cache_test.go | 115 +++++++++++++++++++++++++++++++++- 1 file changed, 113 insertions(+), 2 deletions(-) diff --git a/flag_definition_cache_test.go b/flag_definition_cache_test.go index f3115bc..9ee93c6 100644 --- a/flag_definition_cache_test.go +++ b/flag_definition_cache_test.go @@ -8,6 +8,7 @@ import ( "net/http/httptest" "strings" "sync" + "sync/atomic" "testing" "time" @@ -52,6 +53,7 @@ type fakeFlagDefinitionCache struct { getErr error publishErr error shutdownErr error + onShouldFetch func() onShutdown func(ctx context.Context) error shouldFetchCalls int @@ -62,9 +64,15 @@ type fakeFlagDefinitionCache struct { func (c *fakeFlagDefinitionCache) ShouldFetchFlagDefinitions(context.Context) (bool, error) { c.mu.Lock() - defer c.mu.Unlock() c.shouldFetchCalls++ - return c.shouldFetch, c.shouldFetchErr + 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) { @@ -574,6 +582,109 @@ func TestFlagDefinitionCacheShutdownUsesTheCallerDeadline(t *testing.T) { 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)) From 08e36e0333fd5a2de1eea20eb511cb3892f4dac0 Mon Sep 17 00:00:00 2001 From: Andrey Vasenin Date: Sat, 5 Sep 2026 15:39:17 +0200 Subject: [PATCH 6/9] refactor(flags): rename pollLoopDone to shutdownDone Co-Authored-By: Claude --- featureflags.go | 9 ++++----- flag_definition_cache_test.go | 8 ++++---- 2 files changed, 8 insertions(+), 9 deletions(-) diff --git a/featureflags.go b/featureflags.go index 37465f0..827b921 100644 --- a/featureflags.go +++ b/featureflags.go @@ -85,8 +85,7 @@ type FeatureFlagsPoller struct { firstFeatureFlagRequestFinished chan bool shutdown chan bool forceReload chan bool - // pollLoopDone is closed when the polling goroutine has returned. - pollLoopDone chan struct{} + shutdownDone chan struct{} // state holds all flag-related data using atomic pointer for lock-free reads state atomic.Pointer[flagsState] @@ -480,7 +479,7 @@ func newFeatureFlagsPoller( firstFeatureFlagRequestFinished: make(chan bool), shutdown: make(chan bool), forceReload: make(chan bool), - pollLoopDone: make(chan struct{}), + shutdownDone: make(chan struct{}), personalApiKey: personalApiKey, projectApiKey: projectApiKey, localEvalUrl: localEvalURL, @@ -499,7 +498,7 @@ func newFeatureFlagsPoller( } func (poller *FeatureFlagsPoller) run() { - defer close(poller.pollLoopDone) + defer close(poller.shutdownDone) poller.fetchNewFeatureFlags() close(poller.firstFeatureFlagRequestFinished) @@ -2447,7 +2446,7 @@ func (poller *FeatureFlagsPoller) shutdownPoller(ctx context.Context) { close(poller.shutdown) select { - case <-poller.pollLoopDone: + case <-poller.shutdownDone: case <-ctx.Done(): poller.Logger.Warnf("[FEATURE FLAGS] Polling loop did not stop before the shutdown deadline: %s", ctx.Err()) } diff --git a/flag_definition_cache_test.go b/flag_definition_cache_test.go index 9ee93c6..7d9e780 100644 --- a/flag_definition_cache_test.go +++ b/flag_definition_cache_test.go @@ -151,8 +151,8 @@ func newCachingTestPoller(t *testing.T, serverURL string, provider FlagDefinitio poller.firstFeatureFlagRequestFinished = make(chan bool) close(poller.firstFeatureFlagRequestFinished) - poller.pollLoopDone = make(chan struct{}) - close(poller.pollLoopDone) + poller.shutdownDone = make(chan struct{}) + close(poller.shutdownDone) return poller } @@ -528,7 +528,7 @@ func TestFlagDefinitionCacheShutdownReleasesProvider(t *testing.T) { require.Equal(t, 1, shutdownCalls, "shutdownPoller waits for the provider to be released") select { - case <-poller.pollLoopDone: + case <-poller.shutdownDone: default: t.Fatal("Shutdown ran while the polling loop was still running") } @@ -692,7 +692,7 @@ func TestFlagDefinitionCacheShutdownStopsWaitingOnADeadDeadline(t *testing.T) { poller := newCachingTestPoller(t, server.URL, provider) poller.shutdown = make(chan bool) // A polling loop that never returns. - poller.pollLoopDone = make(chan struct{}) + poller.shutdownDone = make(chan struct{}) ctx, cancel := context.WithCancel(context.Background()) cancel() From 54d88da595f81e5ec486668ab58687f25f2e58ac Mon Sep 17 00:00:00 2001 From: Andrey Vasenin Date: Sat, 5 Sep 2026 15:42:09 +0200 Subject: [PATCH 7/9] refactor(examples): extract the instance count and drop redundant comments Co-Authored-By: Claude --- examples/flag_definition_cache.go | 24 ++++++++++-------------- 1 file changed, 10 insertions(+), 14 deletions(-) diff --git a/examples/flag_definition_cache.go b/examples/flag_definition_cache.go index b5c29c9..06a2609 100644 --- a/examples/flag_definition_cache.go +++ b/examples/flag_definition_cache.go @@ -23,6 +23,7 @@ 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 @@ -34,7 +35,7 @@ type FileFlagCache struct { } // NewFileFlagCache creates a provider for the given cache directory. Instances that -// should share definitions pass the same directory and name. +// 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 @@ -56,18 +57,14 @@ func (c *FileFlagCache) ShouldFetchFlagDefinitions(_ context.Context) (bool, err return false, err } - switch { - case holder == c.instanceID: - // Already the leader: extend the lock. - case holder == "": - case time.Now().Before(expiry): + heldByAnother := holder != "" && holder != c.instanceID + if heldByAnother && time.Now().Before(expiry) { return false, nil } return true, c.writeLock() } -// GetFlagDefinitions returns the published definitions, or nil when there are none. func (c *FileFlagCache) GetFlagDefinitions(_ context.Context) (*posthog.FlagDefinitionCacheData, error) { contents, err := os.ReadFile(c.cachePath()) if errors.Is(err, os.ErrNotExist) { @@ -109,7 +106,6 @@ func (c *FileFlagCache) OnFlagDefinitionsReceived(_ context.Context, data postho return os.Rename(tmp.Name(), c.cachePath()) } -// Shutdown releases the lock if this instance holds it. func (c *FileFlagCache) Shutdown(_ context.Context) error { holder, _, err := c.readLock() if err != nil || holder != c.instanceID { @@ -147,7 +143,7 @@ func (c *FileFlagCache) writeLock() error { func (c *FileFlagCache) shortID() string { return c.instanceID[:8] } -// TestFlagDefinitionCache runs two clients that share one flag definition cache. +// 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 { @@ -156,12 +152,12 @@ func TestFlagDefinitionCache(projectAPIKey, secretKey, endpoint string) { } defer os.RemoveAll(dir) - fmt.Printf("Two instances sharing the cache in %s\n\n", dir) + fmt.Printf("%d instances sharing the cache in %s\n\n", flagCacheInstances, dir) - clients := make([]posthog.Client, 0, 2) - caches := make([]*FileFlagCache, 0, 2) + clients := make([]posthog.Client, 0, flagCacheInstances) + caches := make([]*FileFlagCache, 0, flagCacheInstances) - for i := 0; i < 2; i++ { + 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) @@ -194,7 +190,7 @@ func TestFlagDefinitionCache(projectAPIKey, secretKey, endpoint string) { time.Sleep(2 * flagCachePollInterval) - fmt.Println("\nEvaluating a flag on both instances, entirely from the shared definitions:") + 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", From 4496825c8d5e92f6a5ff2ea7e5cf4bf618fee138 Mon Sep 17 00:00:00 2001 From: Andrey Vasenin Date: Sat, 5 Sep 2026 15:44:37 +0200 Subject: [PATCH 8/9] style(flags): use chan bool for shutdownDone like the other poller channels Co-Authored-By: Claude --- featureflags.go | 4 ++-- flag_definition_cache_test.go | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/featureflags.go b/featureflags.go index 827b921..d5b4841 100644 --- a/featureflags.go +++ b/featureflags.go @@ -85,7 +85,7 @@ type FeatureFlagsPoller struct { firstFeatureFlagRequestFinished chan bool shutdown chan bool forceReload chan bool - shutdownDone chan struct{} + shutdownDone chan bool // state holds all flag-related data using atomic pointer for lock-free reads state atomic.Pointer[flagsState] @@ -479,7 +479,7 @@ func newFeatureFlagsPoller( firstFeatureFlagRequestFinished: make(chan bool), shutdown: make(chan bool), forceReload: make(chan bool), - shutdownDone: make(chan struct{}), + shutdownDone: make(chan bool), personalApiKey: personalApiKey, projectApiKey: projectApiKey, localEvalUrl: localEvalURL, diff --git a/flag_definition_cache_test.go b/flag_definition_cache_test.go index 7d9e780..0a072af 100644 --- a/flag_definition_cache_test.go +++ b/flag_definition_cache_test.go @@ -151,7 +151,7 @@ func newCachingTestPoller(t *testing.T, serverURL string, provider FlagDefinitio poller.firstFeatureFlagRequestFinished = make(chan bool) close(poller.firstFeatureFlagRequestFinished) - poller.shutdownDone = make(chan struct{}) + poller.shutdownDone = make(chan bool) close(poller.shutdownDone) return poller @@ -692,7 +692,7 @@ func TestFlagDefinitionCacheShutdownStopsWaitingOnADeadDeadline(t *testing.T) { poller := newCachingTestPoller(t, server.URL, provider) poller.shutdown = make(chan bool) // A polling loop that never returns. - poller.shutdownDone = make(chan struct{}) + poller.shutdownDone = make(chan bool) ctx, cancel := context.WithCancel(context.Background()) cancel() From 72cd91d544f419709f7c096b518ca58dab048ec7 Mon Sep 17 00:00:00 2001 From: Andrey Vasenin Date: Sat, 5 Sep 2026 15:46:49 +0200 Subject: [PATCH 9/9] style(flags): declare shutdownDone next to shutdown Co-Authored-By: Claude --- featureflags.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/featureflags.go b/featureflags.go index d5b4841..93e85a9 100644 --- a/featureflags.go +++ b/featureflags.go @@ -84,8 +84,8 @@ type FeatureFlagsPoller struct { // After the request the channel get closed. firstFeatureFlagRequestFinished chan bool shutdown chan bool - forceReload chan bool shutdownDone chan bool + forceReload chan bool // state holds all flag-related data using atomic pointer for lock-free reads state atomic.Pointer[flagsState] @@ -478,8 +478,8 @@ func newFeatureFlagsPoller( poller := FeatureFlagsPoller{ firstFeatureFlagRequestFinished: make(chan bool), shutdown: make(chan bool), - forceReload: make(chan bool), shutdownDone: make(chan bool), + forceReload: make(chan bool), personalApiKey: personalApiKey, projectApiKey: projectApiKey, localEvalUrl: localEvalURL,