From 5433c04a6dc6bfb6df5a87b252a372b2e09b018a Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Tue, 7 Jul 2026 14:12:12 -0700 Subject: [PATCH 1/8] refactor(internal): extract shared file data loading into internal/filedata --- internal/filedata/document.go | 71 ++++++++ internal/filedata/filedata_test.go | 158 +++++++++++++++++ internal/filedata/merge.go | 96 +++++++++++ ldfiledata/file_data_source_impl.go | 140 ++------------- ldfiledatav2/file_data_source_impl.go | 239 +++++--------------------- 5 files changed, 381 insertions(+), 323 deletions(-) create mode 100644 internal/filedata/document.go create mode 100644 internal/filedata/filedata_test.go create mode 100644 internal/filedata/merge.go diff --git a/internal/filedata/document.go b/internal/filedata/document.go new file mode 100644 index 00000000..4c4f9a34 --- /dev/null +++ b/internal/filedata/document.go @@ -0,0 +1,71 @@ +// Package filedata contains the file reading, parsing, and merging logic shared by the +// components that load flag and segment data from local files. +package filedata + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "unicode" + + "github.com/launchdarkly/go-sdk-common/v3/ldvalue" + "github.com/launchdarkly/go-server-sdk-evaluation/v3/ldbuilders" + "github.com/launchdarkly/go-server-sdk-evaluation/v3/ldmodel" + + "gopkg.in/ghodss/yaml.v1" +) + +// Document is the parsed form of a single data file. A document may contain full flag +// definitions, simplified flag-key-to-value entries, and segment definitions. +type Document struct { + Flags *map[string]ldmodel.FeatureFlag + FlagValues *map[string]ldvalue.Value + Segments *map[string]ldmodel.Segment +} + +// ReadFile reads and parses a single data file, which may be in JSON or YAML format. +func ReadFile(path string) (Document, error) { + var data Document + var rawData []byte + var err error + if rawData, err = os.ReadFile(path); err != nil { //nolint:gosec // G304: ok to read file into variable + return data, fmt.Errorf("unable to read file: %s", err) + } + if detectJSON(rawData) { + err = json.Unmarshal(rawData, &data) + } else { + err = yaml.Unmarshal(rawData, &data) + } + if err != nil { + err = fmt.Errorf("error parsing file: %s", err) + } + return data, err +} + +func detectJSON(rawData []byte) bool { + // A valid JSON file for our purposes must be an object, i.e. it must start with '{' + return strings.HasPrefix(strings.TrimLeftFunc(string(rawData), unicode.IsSpace), "{") +} + +// AbsFilePaths converts each of the given paths to an absolute path. +func AbsFilePaths(paths []string) ([]string, error) { + absPaths := make([]string, 0) + for _, p := range paths { + absPath, err := filepath.Abs(p) + if err != nil { + // COVERAGE: there's no reliable cross-platform way to simulate an invalid path in unit tests + return nil, fmt.Errorf("unable to determine absolute path for '%s'", p) + } + absPaths = append(absPaths, absPath) + } + return absPaths, nil +} + +// MakeFlagWithValue expands a flag-key-to-value entry into a full flag definition that +// returns the given value for every context. +func MakeFlagWithValue(key string, v interface{}) *ldmodel.FeatureFlag { + flag := ldbuilders.NewFlagBuilder(key).SingleVariation(ldvalue.CopyArbitraryValue(v)).Build() + return &flag +} diff --git a/internal/filedata/filedata_test.go b/internal/filedata/filedata_test.go new file mode 100644 index 00000000..01126c55 --- /dev/null +++ b/internal/filedata/filedata_test.go @@ -0,0 +1,158 @@ +package filedata + +import ( + "os" + "path/filepath" + "testing" + + "github.com/launchdarkly/go-sdk-common/v3/ldvalue" + "github.com/launchdarkly/go-server-sdk-evaluation/v3/ldbuilders" + "github.com/launchdarkly/go-server-sdk-evaluation/v3/ldmodel" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func writeTempFile(t *testing.T, content string) string { + t.Helper() + f, err := os.CreateTemp(t.TempDir(), "filedata-test") + require.NoError(t, err) + _, err = f.WriteString(content) + require.NoError(t, err) + require.NoError(t, f.Close()) + return f.Name() +} + +func TestReadFileJSON(t *testing.T) { + path := writeTempFile(t, `{"flagValues": {"my-flag": true}, "segments": {"my-segment": {"key": "my-segment", "version": 3}}}`) + doc, err := ReadFile(path) + require.NoError(t, err) + require.NotNil(t, doc.FlagValues) + assert.Equal(t, ldvalue.Bool(true), (*doc.FlagValues)["my-flag"]) + require.NotNil(t, doc.Segments) + assert.Equal(t, 3, (*doc.Segments)["my-segment"].Version) + assert.Nil(t, doc.Flags) +} + +func TestReadFileYAML(t *testing.T) { + path := writeTempFile(t, "flagValues:\n my-flag: yes\n") + doc, err := ReadFile(path) + require.NoError(t, err) + require.NotNil(t, doc.FlagValues) + assert.Equal(t, ldvalue.Bool(true), (*doc.FlagValues)["my-flag"]) +} + +func TestReadFileErrors(t *testing.T) { + _, err := ReadFile(filepath.Join(t.TempDir(), "nonexistent")) + assert.ErrorContains(t, err, "unable to read file") + + path := writeTempFile(t, `{"flagValues"`) + _, err = ReadFile(path) + assert.ErrorContains(t, err, "error parsing file") + + path = writeTempFile(t, "\t: not yaml") + _, err = ReadFile(path) + assert.ErrorContains(t, err, "error parsing file") +} + +func TestAbsFilePaths(t *testing.T) { + abs, err := AbsFilePaths([]string{"relative/path", string(filepath.Separator) + "already-absolute"}) + require.NoError(t, err) + require.Len(t, abs, 2) + for _, p := range abs { + assert.True(t, filepath.IsAbs(p), "expected absolute path, got %s", p) + } +} + +func TestMakeFlagWithValue(t *testing.T) { + flag := MakeFlagWithValue("my-flag", "on") + assert.Equal(t, "my-flag", flag.Key) + require.Len(t, flag.Variations, 1) + assert.Equal(t, ldvalue.String("on"), flag.Variations[0]) + require.NotNil(t, flag.OffVariation) +} + +func docWithFlag(flag ldmodel.FeatureFlag) Document { + flags := map[string]ldmodel.FeatureFlag{flag.Key: flag} + return Document{Flags: &flags} +} + +func docWithFlagValue(key string, value ldvalue.Value) Document { + values := map[string]ldvalue.Value{key: value} + return Document{FlagValues: &values} +} + +func docWithSegment(segment ldmodel.Segment) Document { + segments := map[string]ldmodel.Segment{segment.Key: segment} + return Document{Segments: &segments} +} + +func TestMergeCombinesDocuments(t *testing.T) { + flag1 := ldbuilders.NewFlagBuilder("flag1").Version(2).Build() + segment1 := ldbuilders.NewSegmentBuilder("segment1").Version(4).Build() + + result, err := Merge(DuplicateKeysFail, + docWithFlag(flag1), docWithFlagValue("flag2", ldvalue.Bool(true)), docWithSegment(segment1)) + require.NoError(t, err) + + require.Len(t, result.Flags, 2) + assert.Equal(t, "flag1", result.Flags[0].Key) + assert.Equal(t, 2, result.Flags[0].Item.Version) + require.IsType(t, &ldmodel.FeatureFlag{}, result.Flags[0].Item.Item) + assert.Equal(t, "flag2", result.Flags[1].Key) + expanded := result.Flags[1].Item.Item.(*ldmodel.FeatureFlag) + require.Len(t, expanded.Variations, 1) + assert.Equal(t, ldvalue.Bool(true), expanded.Variations[0]) + + require.Len(t, result.Segments, 1) + assert.Equal(t, "segment1", result.Segments[0].Key) + assert.Equal(t, 4, result.Segments[0].Item.Version) +} + +func TestMergeDuplicateKeys(t *testing.T) { + flagA := ldbuilders.NewFlagBuilder("flag1").Version(1).Build() + flagB := ldbuilders.NewFlagBuilder("flag1").Version(2).Build() + + t.Run("fail", func(t *testing.T) { + _, err := Merge(DuplicateKeysFail, docWithFlag(flagA), docWithFlag(flagB)) + assert.ErrorContains(t, err, "flag 'flag1' is specified by multiple files") + }) + + t.Run("unrecognized handling behaves as fail", func(t *testing.T) { + _, err := Merge(DuplicateKeysHandling("bogus"), docWithFlag(flagA), docWithFlag(flagB)) + assert.Error(t, err) + }) + + t.Run("ignore all but first", func(t *testing.T) { + result, err := Merge(DuplicateKeysIgnoreAllButFirst, docWithFlag(flagA), docWithFlag(flagB)) + require.NoError(t, err) + require.Len(t, result.Flags, 1) + assert.Equal(t, 1, result.Flags[0].Item.Version) + }) + + t.Run("full flag and flag value collide", func(t *testing.T) { + _, err := Merge(DuplicateKeysFail, docWithFlag(flagA), docWithFlagValue("flag1", ldvalue.Bool(true))) + assert.Error(t, err) + }) + + t.Run("duplicate segments", func(t *testing.T) { + segment := ldbuilders.NewSegmentBuilder("segment1").Build() + _, err := Merge(DuplicateKeysFail, docWithSegment(segment), docWithSegment(segment)) + assert.ErrorContains(t, err, "segment 'segment1' is specified by multiple files") + }) +} + +func TestMergePreservesDocumentOrder(t *testing.T) { + docs := make([]Document, 0, 5) + expectedKeys := []string{"flag-a", "flag-b", "flag-c", "flag-d", "flag-e"} + for _, key := range expectedKeys { + docs = append(docs, docWithFlag(ldbuilders.NewFlagBuilder(key).Build())) + } + result, err := Merge(DuplicateKeysFail, docs...) + require.NoError(t, err) + keys := make([]string, 0, len(result.Flags)) + for _, item := range result.Flags { + keys = append(keys, item.Key) + } + assert.Equal(t, expectedKeys, keys) +} diff --git a/internal/filedata/merge.go b/internal/filedata/merge.go new file mode 100644 index 00000000..302ed225 --- /dev/null +++ b/internal/filedata/merge.go @@ -0,0 +1,96 @@ +package filedata + +import ( + "fmt" + + "github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoretypes" +) + +// DuplicateKeysHandling determines what happens when the same flag or segment key appears +// in more than one document. +// +// The values match the public option types in the packages that expose this behavior, so +// those types can be converted to this one directly. +type DuplicateKeysHandling string + +const ( + // DuplicateKeysFail means a duplicated key causes the merge to fail. + DuplicateKeysFail DuplicateKeysHandling = "fail" + // DuplicateKeysIgnoreAllButFirst means only the first occurrence of a duplicated key is + // used, in the order the documents were given. + DuplicateKeysIgnoreAllButFirst DuplicateKeysHandling = "ignore" +) + +// MergeResult holds the merged items from one or more documents. Items appear in the order +// they were first encountered, processing documents in the order they were given. Item +// values are *ldmodel.FeatureFlag or *ldmodel.Segment. +type MergeResult struct { + Flags []ldstoretypes.KeyedItemDescriptor + Segments []ldstoretypes.KeyedItemDescriptor +} + +type itemCategory string + +const ( + flagCategory itemCategory = "flag" + segmentCategory itemCategory = "segment" +) + +// Merge combines the items of the given documents, expanding flag-value entries into full +// flag definitions and applying the given duplicate-key handling. An unrecognized +// DuplicateKeysHandling value behaves as DuplicateKeysFail. +func Merge(duplicateKeysHandling DuplicateKeysHandling, docs ...Document) (MergeResult, error) { + var result MergeResult + seenKeys := map[itemCategory]map[string]bool{ + flagCategory: {}, + segmentCategory: {}, + } + + insert := func( + items *[]ldstoretypes.KeyedItemDescriptor, + category itemCategory, + key string, + data ldstoretypes.ItemDescriptor, + ) error { + if seenKeys[category][key] { + switch duplicateKeysHandling { + case DuplicateKeysIgnoreAllButFirst: + return nil + default: + return fmt.Errorf("%s '%s' is specified by multiple files", category, key) + } + } + *items = append(*items, ldstoretypes.KeyedItemDescriptor{Key: key, Item: data}) + seenKeys[category][key] = true + return nil + } + + for _, d := range docs { + if d.Flags != nil { + for key, f := range *d.Flags { + data := ldstoretypes.ItemDescriptor{Version: f.Version, Item: &f} + if err := insert(&result.Flags, flagCategory, key, data); err != nil { + return MergeResult{}, err + } + } + } + if d.FlagValues != nil { + for key, value := range *d.FlagValues { + flag := MakeFlagWithValue(key, value) + data := ldstoretypes.ItemDescriptor{Version: flag.Version, Item: flag} + if err := insert(&result.Flags, flagCategory, key, data); err != nil { + return MergeResult{}, err + } + } + } + if d.Segments != nil { + for key, s := range *d.Segments { + data := ldstoretypes.ItemDescriptor{Version: s.Version, Item: &s} + if err := insert(&result.Segments, segmentCategory, key, data); err != nil { + return MergeResult{}, err + } + } + } + } + return result, nil +} diff --git a/ldfiledata/file_data_source_impl.go b/ldfiledata/file_data_source_impl.go index 5f277548..1689028b 100644 --- a/ldfiledata/file_data_source_impl.go +++ b/ldfiledata/file_data_source_impl.go @@ -1,25 +1,15 @@ package ldfiledata import ( - "encoding/json" - "fmt" - "os" - "path/filepath" - "strings" "sync" "time" - "unicode" "github.com/launchdarkly/go-sdk-common/v3/ldlog" - "github.com/launchdarkly/go-sdk-common/v3/ldvalue" - "github.com/launchdarkly/go-server-sdk-evaluation/v3/ldbuilders" - "github.com/launchdarkly/go-server-sdk-evaluation/v3/ldmodel" "github.com/launchdarkly/go-server-sdk/v7/interfaces" "github.com/launchdarkly/go-server-sdk/v7/internal/datakinds" + "github.com/launchdarkly/go-server-sdk/v7/internal/filedata" "github.com/launchdarkly/go-server-sdk/v7/subsystems" "github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoretypes" - - "gopkg.in/ghodss/yaml.v1" ) type fileDataSource struct { @@ -42,7 +32,7 @@ func newFileDataSourceImpl( duplicateKeysHandling DuplicateKeysHandling, reloaderFactory ReloaderFactory, ) (subsystems.DataSource, error) { - abs, err := absFilePaths(filePaths) + abs, err := filedata.AbsFilePaths(filePaths) if err != nil { // COVERAGE: there's no reliable cross-platform way to simulate an invalid path in unit tests return nil, err @@ -90,11 +80,11 @@ func (fs *fileDataSource) reload() { if fs.closeReloaderCh != nil { fs.loggers.Info("Reloading flag data after detecting a change") } - filesData := make([]fileData, 0) + docs := make([]filedata.Document, 0) for _, path := range fs.absFilePaths { - data, err := readFile(path) + doc, err := filedata.ReadFile(path) if err == nil { - filesData = append(filesData, data) + docs = append(docs, doc) } else { fs.loggers.Errorf("Unable to load flags: %s [%s]", err, path) fs.dataSourceUpdates.UpdateStatus(interfaces.DataSourceStateInterrupted, @@ -106,8 +96,12 @@ func (fs *fileDataSource) reload() { return } } - storeData, err := mergeFileData(fs.duplicateKeysHandling, filesData...) + merged, err := filedata.Merge(filedata.DuplicateKeysHandling(fs.duplicateKeysHandling), docs...) if err == nil { + storeData := []ldstoretypes.Collection{ + {Kind: datakinds.Features, Items: merged.Flags}, + {Kind: datakinds.Segments, Items: merged.Segments}, + } if fs.dataSourceUpdates.Init(storeData) { fs.signalStartComplete(true) fs.dataSourceUpdates.UpdateStatus(interfaces.DataSourceStateValid, interfaces.DataSourceErrorInfo{}) @@ -134,120 +128,6 @@ func (fs *fileDataSource) signalStartComplete(succeeded bool) { }) } -func absFilePaths(paths []string) ([]string, error) { - absPaths := make([]string, 0) - for _, p := range paths { - absPath, err := filepath.Abs(p) - if err != nil { - // COVERAGE: there's no reliable cross-platform way to simulate an invalid path in unit tests - return nil, fmt.Errorf("unable to determine absolute path for '%s'", p) - } - absPaths = append(absPaths, absPath) - } - return absPaths, nil -} - -type fileData struct { - Flags *map[string]ldmodel.FeatureFlag - FlagValues *map[string]ldvalue.Value - Segments *map[string]ldmodel.Segment -} - -func insertData( - all map[ldstoretypes.DataKind]map[string]ldstoretypes.ItemDescriptor, - kind ldstoretypes.DataKind, - key string, - data ldstoretypes.ItemDescriptor, - duplicateKeysHandling DuplicateKeysHandling, -) error { - if _, exists := all[kind][key]; exists { - switch duplicateKeysHandling { - case DuplicateKeysIgnoreAllButFirst: - return nil - default: - return fmt.Errorf("%s '%s' is specified by multiple files", kind, key) - } - } - all[kind][key] = data - return nil -} - -func readFile(path string) (fileData, error) { - var data fileData - var rawData []byte - var err error - if rawData, err = os.ReadFile(path); err != nil { //nolint:gosec // G304: ok to read file into variable - return data, fmt.Errorf("unable to read file: %s", err) - } - if detectJSON(rawData) { - err = json.Unmarshal(rawData, &data) - } else { - err = yaml.Unmarshal(rawData, &data) - } - if err != nil { - err = fmt.Errorf("error parsing file: %s", err) - } - return data, err -} - -func detectJSON(rawData []byte) bool { - // A valid JSON file for our purposes must be an object, i.e. it must start with '{' - return strings.HasPrefix(strings.TrimLeftFunc(string(rawData), unicode.IsSpace), "{") -} - -func mergeFileData( - duplicateKeysHandling DuplicateKeysHandling, - allFileData ...fileData, -) ([]ldstoretypes.Collection, error) { - all := map[ldstoretypes.DataKind]map[string]ldstoretypes.ItemDescriptor{ - datakinds.Features: {}, - datakinds.Segments: {}, - } - for _, d := range allFileData { - if d.Flags != nil { - for key, f := range *d.Flags { - ff := f - data := ldstoretypes.ItemDescriptor{Version: f.Version, Item: &ff} - if err := insertData(all, datakinds.Features, key, data, duplicateKeysHandling); err != nil { - return nil, err - } - } - } - if d.FlagValues != nil { - for key, value := range *d.FlagValues { - flag := makeFlagWithValue(key, value) - data := ldstoretypes.ItemDescriptor{Version: flag.Version, Item: flag} - if err := insertData(all, datakinds.Features, key, data, duplicateKeysHandling); err != nil { - return nil, err - } - } - } - if d.Segments != nil { - for key, s := range *d.Segments { - ss := s - data := ldstoretypes.ItemDescriptor{Version: s.Version, Item: &ss} - if err := insertData(all, datakinds.Segments, key, data, duplicateKeysHandling); err != nil { - return nil, err - } - } - } - } - ret := []ldstoretypes.Collection{} - for kind, itemsMap := range all { - items := make([]ldstoretypes.KeyedItemDescriptor, 0, len(itemsMap)) - for k, v := range itemsMap { - items = append(items, ldstoretypes.KeyedItemDescriptor{Key: k, Item: v}) - } - ret = append(ret, ldstoretypes.Collection{Kind: kind, Items: items}) - } - return ret, nil -} - -func makeFlagWithValue(key string, v interface{}) *ldmodel.FeatureFlag { - flag := ldbuilders.NewFlagBuilder(key).SingleVariation(ldvalue.CopyArbitraryValue(v)).Build() - return &flag -} - // Close is called automatically when the client is closed. func (fs *fileDataSource) Close() (err error) { fs.closeOnce.Do(func() { diff --git a/ldfiledatav2/file_data_source_impl.go b/ldfiledatav2/file_data_source_impl.go index fb177d55..0f988b5e 100644 --- a/ldfiledatav2/file_data_source_impl.go +++ b/ldfiledatav2/file_data_source_impl.go @@ -2,28 +2,18 @@ package ldfiledatav2 import ( "context" - "encoding/json" "errors" "fmt" - "os" - "path/filepath" - "strings" - "sync" "sync/atomic" "time" - "unicode" "github.com/launchdarkly/go-sdk-common/v3/ldlog" - "github.com/launchdarkly/go-sdk-common/v3/ldvalue" - "github.com/launchdarkly/go-server-sdk-evaluation/v3/ldbuilders" - "github.com/launchdarkly/go-server-sdk-evaluation/v3/ldmodel" "github.com/launchdarkly/go-server-sdk/v7/interfaces" "github.com/launchdarkly/go-server-sdk/v7/internal" + "github.com/launchdarkly/go-server-sdk/v7/internal/filedata" "github.com/launchdarkly/go-server-sdk/v7/subsystems" "github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoreimpl" "github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoretypes" - - "gopkg.in/ghodss/yaml.v1" ) type fileDataSource struct { @@ -49,7 +39,7 @@ func newFileDataSourceImpl( duplicateKeysHandling DuplicateKeysHandling, reloaderFactory ReloaderFactory, ) (subsystems.DataSynchronizer, error) { - abs, err := absFilePaths(filePaths) + abs, err := filedata.AbsFilePaths(filePaths) if err != nil { // COVERAGE: there's no reliable cross-platform way to simulate an invalid path in unit tests return nil, err @@ -91,6 +81,7 @@ func (fs *fileDataSource) Sync(ds subsystems.DataSelector) <-chan subsystems.Dat fs.reload() if fs.reloaderFactory != nil { + fs.closeReloaderCh = make(chan struct{}) err := fs.reloaderFactory(fs.absFilePaths, fs.loggers, fs.reload, fs.closeReloaderCh) if err != nil { fs.loggers.Errorf("Unable to start reloader: %s\n", err) @@ -136,49 +127,14 @@ func (fs *fileDataSource) Sync(ds subsystems.DataSelector) <-chan subsystems.Dat } func (fs *fileDataSource) Fetch(ds subsystems.DataSelector, ctx context.Context) (*subsystems.Basis, bool, error) { - changeSetChan := fs.changeSetBroadcaster.AddListener() - statusChan := fs.statusBroadcaster.AddListener() - - changeset := subsystems.NewChangeSetBuilder().NoChanges() - - var err error - basis := &subsystems.Basis{ - ChangeSet: *changeset, - Persist: false, + changeSet, err := fs.load() + if err != nil { + return nil, false, err } - - var wg sync.WaitGroup - wg.Add(1) - - go func() { - fs.reload() - - defer wg.Done() - - select { - case changeSet, ok := <-changeSetChan: - if !ok { - return - } - basis.ChangeSet = changeSet - return - case statusChange, ok := <-statusChan: - if !ok { - return - } - - if statusChange.State != interfaces.DataSourceStateValid { - err = errors.New("data source did not receive change set") - return - } - default: - return - } - }() - - wg.Wait() - - return basis, false, err + return &subsystems.Basis{ + ChangeSet: *changeSet, + Persist: false, + }, false, nil } // Reload tells the data source to immediately attempt to reread all of the configured source files @@ -189,180 +145,82 @@ func (fs *fileDataSource) reload() { fs.loggers.Info("Reloading flag data after detecting a change") } - filesData := make([]fileData, 0) - for _, path := range fs.absFilePaths { - data, err := readFile(path) - if err == nil { - filesData = append(filesData, data) - } else { - fs.loggers.Errorf("Unable to load flags: %s [%s]", err, path) - fs.statusBroadcaster.Broadcast(interfaces.DataSynchronizerStatus{ - State: interfaces.DataSourceStateInterrupted, - Error: interfaces.DataSourceErrorInfo{ - Kind: interfaces.DataSourceErrorKindUnknown, - StatusCode: 0, - Message: err.Error(), - Time: time.Time{}, - }, - }) - return - } - } - - fs.version++ - changeSet, err := mergeFileData(fs.duplicateKeysHandling, fs.version, filesData...) - + changeSet, err := fs.load() if err == nil { fs.changeSetBroadcaster.Broadcast(*changeSet) } else { + fs.loggers.Errorf("Unable to load flags: %s", err) + errorKind := interfaces.DataSourceErrorKindInvalidData + var readErr *fileReadError + if errors.As(err, &readErr) { + errorKind = interfaces.DataSourceErrorKindUnknown + } fs.statusBroadcaster.Broadcast(interfaces.DataSynchronizerStatus{ State: interfaces.DataSourceStateInterrupted, Error: interfaces.DataSourceErrorInfo{ - Kind: interfaces.DataSourceErrorKindInvalidData, + Kind: errorKind, StatusCode: 0, Message: err.Error(), Time: time.Time{}, }, FallbackToFDv1: false, }) - fs.loggers.Error(err) } } -func absFilePaths(paths []string) ([]string, error) { - absPaths := make([]string, 0) - for _, p := range paths { - absPath, err := filepath.Abs(p) - if err != nil { - // COVERAGE: there's no reliable cross-platform way to simulate an invalid path in unit tests - return nil, fmt.Errorf("unable to determine absolute path for '%s'", p) - } - absPaths = append(absPaths, absPath) - } - return absPaths, nil +// fileReadError distinguishes a failure to read or parse one of the source files from a +// failure to merge their contents. +type fileReadError struct { + err error + path string } -type fileData struct { - Flags *map[string]ldmodel.FeatureFlag - FlagValues *map[string]ldvalue.Value - Segments *map[string]ldmodel.Segment +func (e *fileReadError) Error() string { + return fmt.Sprintf("%s [%s]", e.err, e.path) } -func insertDataIntoCollection( - items *[]ldstoretypes.KeyedItemDescriptor, - seenKeys map[subsystems.ObjectKind]map[string]bool, - objectKind subsystems.ObjectKind, - key string, - data ldstoretypes.ItemDescriptor, - duplicateKeysHandling DuplicateKeysHandling, -) error { - if _, exists := seenKeys[objectKind][key]; exists { - switch duplicateKeysHandling { - case DuplicateKeysIgnoreAllButFirst: - return nil - default: - return fmt.Errorf("%s '%s' is specified by multiple files", objectKind, key) - } - } - - *items = append(*items, ldstoretypes.KeyedItemDescriptor{ - Key: key, - Item: data, - }) - seenKeys[objectKind][key] = true - - return nil +func (e *fileReadError) Unwrap() error { + return e.err } -func readFile(path string) (fileData, error) { - var data fileData - var rawData []byte - var err error - if rawData, err = os.ReadFile(path); err != nil { //nolint:gosec // G304: ok to read file into variable - return data, fmt.Errorf("unable to read file: %s", err) - } - if detectJSON(rawData) { - err = json.Unmarshal(rawData, &data) - } else { - err = yaml.Unmarshal(rawData, &data) +// load synchronously reads and merges all of the configured source files, returning the +// result as a full-transfer change set. +func (fs *fileDataSource) load() (*subsystems.ChangeSet, error) { + docs := make([]filedata.Document, 0) + for _, path := range fs.absFilePaths { + doc, err := filedata.ReadFile(path) + if err != nil { + return nil, &fileReadError{err: err, path: path} + } + docs = append(docs, doc) } + + merged, err := filedata.Merge(filedata.DuplicateKeysHandling(fs.duplicateKeysHandling), docs...) if err != nil { - err = fmt.Errorf("error parsing file: %s", err) + return nil, err } - return data, err -} - -func detectJSON(rawData []byte) bool { - // A valid JSON file for our purposes must be an object, i.e. it must start with '{' - return strings.HasPrefix(strings.TrimLeftFunc(string(rawData), unicode.IsSpace), "{") -} -func mergeFileData( - duplicateKeysHandling DuplicateKeysHandling, - version int, - allFileData ...fileData, -) (*subsystems.ChangeSet, error) { + fs.version++ intent := subsystems.ServerIntent{ Payload: subsystems.Payload{ ID: "", - Target: version, + Target: fs.version, Code: subsystems.IntentTransferFull, Reason: "payload-missing", }, } - // Build collections directly instead of using ChangeSetBuilder - flagItems := make([]ldstoretypes.KeyedItemDescriptor, 0) - segmentItems := make([]ldstoretypes.KeyedItemDescriptor, 0) - - seenKeys := map[subsystems.ObjectKind]map[string]bool{ - subsystems.FlagKind: {}, - subsystems.SegmentKind: {}, - } - - for _, d := range allFileData { - if d.Flags != nil { - for key, f := range *d.Flags { - data := ldstoretypes.ItemDescriptor{Version: f.Version, Item: &f} - err := insertDataIntoCollection(&flagItems, seenKeys, subsystems.FlagKind, key, data, duplicateKeysHandling) - if err != nil { - return nil, err - } - } - } - if d.FlagValues != nil { - for key, value := range *d.FlagValues { - flag := makeFlagWithValue(key, value) - data := ldstoretypes.ItemDescriptor{Version: flag.Version, Item: flag} - err := insertDataIntoCollection(&flagItems, seenKeys, subsystems.FlagKind, key, data, duplicateKeysHandling) - if err != nil { - return nil, err - } - } - } - if d.Segments != nil { - for key, s := range *d.Segments { - data := ldstoretypes.ItemDescriptor{Version: s.Version, Item: &s} - err := insertDataIntoCollection(&segmentItems, seenKeys, subsystems.SegmentKind, key, data, duplicateKeysHandling) - if err != nil { - return nil, err - } - } - } - } - - // Build collections collections := make([]ldstoretypes.Collection, 0, 2) - if len(flagItems) > 0 { + if len(merged.Flags) > 0 { collections = append(collections, ldstoretypes.Collection{ Kind: ldstoreimpl.Features(), - Items: flagItems, + Items: merged.Flags, }) } - if len(segmentItems) > 0 { + if len(merged.Segments) > 0 { collections = append(collections, ldstoretypes.Collection{ Kind: ldstoreimpl.Segments(), - Items: segmentItems, + Items: merged.Segments, }) } @@ -373,11 +231,6 @@ func mergeFileData( return subsystems.NewChangeSetFromCollections(intent, subsystems.NoSelector(), collections) } -func makeFlagWithValue(key string, v interface{}) *ldmodel.FeatureFlag { - flag := ldbuilders.NewFlagBuilder(key).SingleVariation(ldvalue.CopyArbitraryValue(v)).Build() - return &flag -} - // Close is called automatically when the client is closed. func (fs *fileDataSource) Close() (err error) { if swapped := fs.closed.CompareAndSwap(false, true); swapped { From 18fc66280a2ec5297718719c383478b02375aaeb Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Tue, 7 Jul 2026 14:20:46 -0700 Subject: [PATCH 2/8] feat(internal): add reload orchestration with debouncing and failure retry to file data loading --- internal/filedata/document.go | 27 ++- internal/filedata/reloader.go | 230 ++++++++++++++++++++++++++ internal/filedata/reloader_test.go | 207 +++++++++++++++++++++++ ldfiledata/file_data_source_impl.go | 86 +++++----- ldfiledatav2/file_data_source_impl.go | 117 +++++++------ ldfiledatav2/file_data_source_test.go | 12 +- 6 files changed, 570 insertions(+), 109 deletions(-) create mode 100644 internal/filedata/reloader.go create mode 100644 internal/filedata/reloader_test.go diff --git a/internal/filedata/document.go b/internal/filedata/document.go index 4c4f9a34..2e250d63 100644 --- a/internal/filedata/document.go +++ b/internal/filedata/document.go @@ -25,14 +25,33 @@ type Document struct { Segments *map[string]ldmodel.Segment } +// ReadError indicates that one of the source files could not be read or parsed. It +// distinguishes a per-file failure from a failure to merge the files' contents. +type ReadError struct { + Err error + Path string +} + +func (e *ReadError) Error() string { + return fmt.Sprintf("%s [%s]", e.Err, e.Path) +} + +func (e *ReadError) Unwrap() error { + return e.Err +} + // ReadFile reads and parses a single data file, which may be in JSON or YAML format. func ReadFile(path string) (Document, error) { + rawData, err := os.ReadFile(path) //nolint:gosec // G304: ok to read file into variable + if err != nil { + return Document{}, fmt.Errorf("unable to read file: %s", err) + } + return parseDocument(rawData) +} + +func parseDocument(rawData []byte) (Document, error) { var data Document - var rawData []byte var err error - if rawData, err = os.ReadFile(path); err != nil { //nolint:gosec // G304: ok to read file into variable - return data, fmt.Errorf("unable to read file: %s", err) - } if detectJSON(rawData) { err = json.Unmarshal(rawData, &data) } else { diff --git a/internal/filedata/reloader.go b/internal/filedata/reloader.go new file mode 100644 index 00000000..fb52ecc9 --- /dev/null +++ b/internal/filedata/reloader.go @@ -0,0 +1,230 @@ +package filedata + +import ( + "bytes" + "crypto/sha256" + "errors" + "os" + "sync" + "time" + + "github.com/launchdarkly/go-sdk-common/v3/ldlog" +) + +const ( + // DefaultDebounceDelay is a settle window long enough to coalesce the burst of change + // notifications produced by a single file edit, and short enough to stay responsive. + DefaultDebounceDelay = 100 * time.Millisecond + // DefaultRetryDelay bounds how long a failed reload can go uncorrected when no further + // change notification arrives, for example when the failure came from reading a file + // mid-write. Reading a local file is cheap, so this can be short. + DefaultRetryDelay = time.Second +) + +// ReloaderConfig configures a Reloader. +type ReloaderConfig struct { + // Paths is the list of files to load, already resolved to absolute paths. The order is + // significant: it determines which file wins under the duplicate-key handling. + Paths []string + // DuplicateKeysHandling determines what happens when the same key appears in more than + // one file. + DuplicateKeysHandling DuplicateKeysHandling + // Loggers receives log output about reloads and failures. + Loggers ldlog.Loggers + // Apply is invoked with each successfully merged result. Calls are serialized on the + // reloader's own goroutine (or, for ReloadNow, under the same lock), so implementations + // do not need their own synchronization against other reloads. + Apply func(MergeResult) + // OnError is invoked for each failed reload. The error is a *ReadError when a file could + // not be read or parsed, or a merge error otherwise. The reloader logs failures itself, + // so implementations only need to update their own state. + OnError func(err error) + // DebounceDelay is how long to wait after a Trigger call for further calls to settle + // before reloading, coalescing bursts of change notifications into one reload. If zero, + // each Trigger reloads immediately. + DebounceDelay time.Duration + // RetryDelay is how long to wait after a failed reload before automatically retrying, + // so that a failure observed while a file was being rewritten recovers even if no + // further change notification arrives. If zero, there is no automatic retry. + RetryDelay time.Duration + // SkipUnchanged, if true, suppresses the Apply call when the files' raw contents are + // byte-identical to the last successfully applied contents. + SkipUnchanged bool +} + +// Reloader owns the reload cycle for a set of data files: it serializes reloads, debounces +// change signals, retains the last good result on failure (by not calling Apply), retries +// after failures, and skips no-op applications. +type Reloader struct { + cfg ReloaderConfig + triggerCh chan struct{} + closeCh chan struct{} + doneCh chan struct{} + closeOnce sync.Once + + // reloadMu serializes the actual load work between ReloadNow and the run goroutine. + reloadMu sync.Mutex + lastGoodHash []byte + lastErrorMsg string +} + +// NewReloader creates a started Reloader. The caller should perform the initial load with +// ReloadNow, route change signals to Trigger, and call Close when finished. +func NewReloader(cfg ReloaderConfig) *Reloader { + r := &Reloader{ + cfg: cfg, + triggerCh: make(chan struct{}, 1), + closeCh: make(chan struct{}), + doneCh: make(chan struct{}), + } + go r.run() + return r +} + +// ReloadNow synchronously loads the files and applies the result (or reports the failure). +// It is used for the initial load; a failure here schedules the same automatic retry as a +// failed triggered reload. +func (r *Reloader) ReloadNow() { + if !r.reload() && r.cfg.RetryDelay > 0 { + r.Trigger() + } +} + +// Trigger signals that the files may have changed and a reload should happen after the +// debounce delay. It never blocks; signals arriving while a reload is already pending are +// coalesced. +func (r *Reloader) Trigger() { + select { + case r.triggerCh <- struct{}{}: + default: + } +} + +// Close stops the reloader. No reload will call Apply or OnError after Close returns. +func (r *Reloader) Close() { + r.closeOnce.Do(func() { + close(r.closeCh) + <-r.doneCh + // An in-flight reload holds reloadMu, so acquiring it guarantees the reload has + // finished before Close returns. + r.reloadMu.Lock() + defer r.reloadMu.Unlock() + }) +} + +func (r *Reloader) run() { + defer close(r.doneCh) + var debounceTimer, retryTimer *time.Timer + var debounceC, retryC <-chan time.Time + stopTimer := func(timer *time.Timer) { + if timer != nil { + timer.Stop() + } + } + defer func() { + stopTimer(debounceTimer) + stopTimer(retryTimer) + }() + + reloadAndMaybeRetry := func(isRetry bool) { + if isRetry { + r.cfg.Loggers.Debug("Retrying flag data load after earlier failure") + } else { + r.cfg.Loggers.Info("Reloading flag data after detecting a change") + } + // A pending retry is superseded by this reload: it either succeeded, or it failed + // and arms a fresh retry below. + stopTimer(retryTimer) + retryTimer, retryC = nil, nil + if !r.reload() && r.cfg.RetryDelay > 0 { + retryTimer = time.NewTimer(r.cfg.RetryDelay) + retryC = retryTimer.C + } + } + + for { + select { + case <-r.closeCh: + return + case <-r.triggerCh: + if r.cfg.DebounceDelay <= 0 { + reloadAndMaybeRetry(false) + continue + } + if debounceTimer == nil { + debounceTimer = time.NewTimer(r.cfg.DebounceDelay) + debounceC = debounceTimer.C + } else { + stopTimer(debounceTimer) + debounceTimer.Reset(r.cfg.DebounceDelay) + } + case <-debounceC: + debounceTimer, debounceC = nil, nil + reloadAndMaybeRetry(false) + case <-retryC: + retryTimer, retryC = nil, nil + reloadAndMaybeRetry(true) + } + } +} + +// reload performs one full load of all configured files, returning true on success (including +// a skipped no-op application). The whole set is re-read on every reload: entries are combined +// across files in order, so a change to one file can alter which file wins for a key. +func (r *Reloader) reload() bool { + r.reloadMu.Lock() + defer r.reloadMu.Unlock() + + // A trigger already queued when Close was called can still reach here; the select in + // run() does not prioritize closeCh over triggerCh. + select { + case <-r.closeCh: + return true + default: + } + + docs := make([]Document, 0, len(r.cfg.Paths)) + hasher := sha256.New() + for _, path := range r.cfg.Paths { + rawData, err := os.ReadFile(path) //nolint:gosec // G304: ok to read file into variable + if err != nil { + return r.fail(&ReadError{Err: errors.New("unable to read file: " + err.Error()), Path: path}) + } + _, _ = hasher.Write(rawData) + _, _ = hasher.Write([]byte{0}) + doc, err := parseDocument(rawData) + if err != nil { + return r.fail(&ReadError{Err: err, Path: path}) + } + docs = append(docs, doc) + } + + merged, err := Merge(r.cfg.DuplicateKeysHandling, docs...) + if err != nil { + return r.fail(err) + } + + r.lastErrorMsg = "" + hash := hasher.Sum(nil) + if r.cfg.SkipUnchanged && bytes.Equal(hash, r.lastGoodHash) { + return true + } + r.lastGoodHash = hash + r.cfg.Apply(merged) + return true +} + +func (r *Reloader) fail(err error) bool { + // With automatic retries, a persistent failure would repeat the same log entry on every + // attempt, so repeats of an identical failure are demoted to debug level. + if err.Error() == r.lastErrorMsg { + r.cfg.Loggers.Debugf("Unable to load flags: %s", err) + } else { + r.lastErrorMsg = err.Error() + r.cfg.Loggers.Errorf("Unable to load flags: %s", err) + } + if r.cfg.OnError != nil { + r.cfg.OnError(err) + } + return false +} diff --git a/internal/filedata/reloader_test.go b/internal/filedata/reloader_test.go new file mode 100644 index 00000000..0dddfba8 --- /dev/null +++ b/internal/filedata/reloader_test.go @@ -0,0 +1,207 @@ +package filedata + +import ( + "os" + "path/filepath" + "testing" + "time" + + "github.com/launchdarkly/go-sdk-common/v3/ldlog" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const testTimeout = 5 * time.Second + +type reloaderFixture struct { + path string + applied chan MergeResult + errored chan error + reloader *Reloader +} + +func newReloaderFixture(t *testing.T, initialContent string, configure func(*ReloaderConfig)) *reloaderFixture { + t.Helper() + f := &reloaderFixture{ + path: filepath.Join(t.TempDir(), "data.json"), + applied: make(chan MergeResult, 100), + errored: make(chan error, 100), + } + f.write(t, initialContent) + cfg := ReloaderConfig{ + Paths: []string{f.path}, + DuplicateKeysHandling: DuplicateKeysFail, + Loggers: ldlog.NewDisabledLoggers(), + Apply: func(result MergeResult) { f.applied <- result }, + OnError: func(err error) { f.errored <- err }, + } + if configure != nil { + configure(&cfg) + } + f.reloader = NewReloader(cfg) + t.Cleanup(f.reloader.Close) + return f +} + +func (f *reloaderFixture) write(t *testing.T, content string) { + t.Helper() + require.NoError(t, os.WriteFile(f.path, []byte(content), 0600)) +} + +func (f *reloaderFixture) requireApplied(t *testing.T) MergeResult { + t.Helper() + select { + case result := <-f.applied: + return result + case <-time.After(testTimeout): + require.FailNow(t, "timed out waiting for Apply") + return MergeResult{} + } +} + +func (f *reloaderFixture) requireErrored(t *testing.T) error { + t.Helper() + select { + case err := <-f.errored: + return err + case <-time.After(testTimeout): + require.FailNow(t, "timed out waiting for OnError") + return nil + } +} + +func (f *reloaderFixture) requireQuiet(t *testing.T, duration time.Duration) { + t.Helper() + select { + case <-f.applied: + require.FailNow(t, "unexpected Apply call") + case <-f.errored: + require.FailNow(t, "unexpected OnError call") + case <-time.After(duration): + } +} + +func TestReloaderInitialLoad(t *testing.T) { + f := newReloaderFixture(t, `{"flagValues": {"flag1": true}}`, nil) + f.reloader.ReloadNow() + result := f.requireApplied(t) + require.Len(t, result.Flags, 1) + assert.Equal(t, "flag1", result.Flags[0].Key) +} + +func TestReloaderReportsFailureAndRetainsNothing(t *testing.T) { + f := newReloaderFixture(t, `{"flagValues"`, nil) + f.reloader.ReloadNow() + err := f.requireErrored(t) + var readErr *ReadError + require.ErrorAs(t, err, &readErr) + assert.Equal(t, f.path, readErr.Path) + f.requireQuiet(t, 50*time.Millisecond) +} + +func TestReloaderDebounceCoalescesTriggers(t *testing.T) { + f := newReloaderFixture(t, `{"flagValues": {"flag1": true}}`, func(cfg *ReloaderConfig) { + cfg.DebounceDelay = 20 * time.Millisecond + }) + f.write(t, `{"flagValues": {"flag1": false}}`) + for i := 0; i < 20; i++ { + f.reloader.Trigger() + time.Sleep(time.Millisecond) + } + f.requireApplied(t) + // All of the triggers, arriving within the settle window, coalesced into one reload. + f.requireQuiet(t, 100*time.Millisecond) +} + +func TestReloaderRetriesAfterFailureWithoutFurtherTriggers(t *testing.T) { + f := newReloaderFixture(t, `{"flagValues": {"flag1": true}}`, func(cfg *ReloaderConfig) { + cfg.RetryDelay = 20 * time.Millisecond + }) + f.reloader.ReloadNow() + f.requireApplied(t) + + f.write(t, `{"flagValues"`) + f.reloader.Trigger() + f.requireErrored(t) + + // Fix the file without triggering; only the automatic retry can observe the fix. + f.write(t, `{"flagValues": {"flag1": false}}`) + f.requireApplied(t) +} + +func TestReloaderStopsRetryingAfterSuccess(t *testing.T) { + f := newReloaderFixture(t, `{"flagValues"`, func(cfg *ReloaderConfig) { + cfg.RetryDelay = 10 * time.Millisecond + cfg.SkipUnchanged = true + }) + f.reloader.ReloadNow() + f.requireErrored(t) + + f.write(t, `{"flagValues": {"flag1": true}}`) + f.requireApplied(t) + + // After the successful reload there are no further attempts (a retry would re-apply or + // re-error; with SkipUnchanged set, a stray retry of identical content stays silent, so + // also verify via a changed file that no reload happens without a trigger). + f.write(t, `{"flagValues": {"flag1": false}}`) + f.requireQuiet(t, 100*time.Millisecond) +} + +func TestReloaderSkipUnchanged(t *testing.T) { + f := newReloaderFixture(t, `{"flagValues": {"flag1": true}}`, func(cfg *ReloaderConfig) { + cfg.SkipUnchanged = true + }) + f.reloader.ReloadNow() + f.requireApplied(t) + + f.reloader.Trigger() + f.requireQuiet(t, 100*time.Millisecond) + + f.write(t, `{"flagValues": {"flag1": false}}`) + f.reloader.Trigger() + f.requireApplied(t) +} + +func TestReloaderAppliesEveryReloadWhenSkipUnchangedIsOff(t *testing.T) { + f := newReloaderFixture(t, `{"flagValues": {"flag1": true}}`, nil) + f.reloader.ReloadNow() + f.requireApplied(t) + f.reloader.Trigger() + f.requireApplied(t) +} + +func TestReloaderMergesMultipleFilesInOrder(t *testing.T) { + dir := t.TempDir() + path1 := filepath.Join(dir, "first.json") + path2 := filepath.Join(dir, "second.json") + require.NoError(t, os.WriteFile(path1, []byte(`{"flagValues": {"flag1": "first"}}`), 0600)) + require.NoError(t, os.WriteFile(path2, []byte(`{"flagValues": {"flag1": "second"}}`), 0600)) + + applied := make(chan MergeResult, 10) + r := NewReloader(ReloaderConfig{ + Paths: []string{path1, path2}, + DuplicateKeysHandling: DuplicateKeysIgnoreAllButFirst, + Loggers: ldlog.NewDisabledLoggers(), + Apply: func(result MergeResult) { applied <- result }, + }) + t.Cleanup(r.Close) + r.ReloadNow() + + select { + case result := <-applied: + require.Len(t, result.Flags, 1) + case <-time.After(testTimeout): + require.FailNow(t, "timed out waiting for Apply") + } +} + +func TestReloaderDoesNothingAfterClose(t *testing.T) { + f := newReloaderFixture(t, `{"flagValues": {"flag1": true}}`, nil) + f.reloader.ReloadNow() + f.requireApplied(t) + f.reloader.Close() + f.reloader.Close() // idempotent + f.reloader.Trigger() + f.requireQuiet(t, 50*time.Millisecond) +} diff --git a/ldfiledata/file_data_source_impl.go b/ldfiledata/file_data_source_impl.go index 1689028b..3227cca3 100644 --- a/ldfiledata/file_data_source_impl.go +++ b/ldfiledata/file_data_source_impl.go @@ -17,6 +17,7 @@ type fileDataSource struct { absFilePaths []string duplicateKeysHandling DuplicateKeysHandling reloaderFactory ReloaderFactory + reloader *filedata.Reloader loggers ldlog.Loggers isInitialized bool readyCh chan<- struct{} @@ -55,7 +56,25 @@ func (fs *fileDataSource) IsInitialized() bool { func (fs *fileDataSource) Start(closeWhenReady chan<- struct{}) { fs.readyCh = closeWhenReady - fs.reload() + + // Debouncing and automatic retries only matter when something can trigger further + // reloads; a source configured without a reloader loads exactly once. + var debounceDelay, retryDelay time.Duration + if fs.reloaderFactory != nil { + debounceDelay = filedata.DefaultDebounceDelay + retryDelay = filedata.DefaultRetryDelay + } + fs.reloader = filedata.NewReloader(filedata.ReloaderConfig{ + Paths: fs.absFilePaths, + DuplicateKeysHandling: filedata.DuplicateKeysHandling(fs.duplicateKeysHandling), + Loggers: fs.loggers, + Apply: fs.applyData, + OnError: fs.handleError, + DebounceDelay: debounceDelay, + RetryDelay: retryDelay, + SkipUnchanged: true, + }) + fs.reloader.ReloadNow() // If there is no reloader, then we signal readiness immediately regardless of whether the // data load succeeded or failed. @@ -65,60 +84,34 @@ func (fs *fileDataSource) Start(closeWhenReady chan<- struct{}) { } // If there is a reloader, and if we haven't yet successfully loaded data, then the - // readiness signal will happen the first time we do get valid data (in reload). + // readiness signal will happen the first time we do get valid data (in applyData). fs.closeReloaderCh = make(chan struct{}) - err := fs.reloaderFactory(fs.absFilePaths, fs.loggers, fs.reload, fs.closeReloaderCh) + err := fs.reloaderFactory(fs.absFilePaths, fs.loggers, fs.reloader.Trigger, fs.closeReloaderCh) if err != nil { fs.loggers.Errorf("Unable to start reloader: %s\n", err) } } -// Reload tells the data source to immediately attempt to reread all of the configured source files -// and update the feature flag state. If any file cannot be loaded or parsed, the flag state will not -// be modified. -func (fs *fileDataSource) reload() { - if fs.closeReloaderCh != nil { - fs.loggers.Info("Reloading flag data after detecting a change") - } - docs := make([]filedata.Document, 0) - for _, path := range fs.absFilePaths { - doc, err := filedata.ReadFile(path) - if err == nil { - docs = append(docs, doc) - } else { - fs.loggers.Errorf("Unable to load flags: %s [%s]", err, path) - fs.dataSourceUpdates.UpdateStatus(interfaces.DataSourceStateInterrupted, - interfaces.DataSourceErrorInfo{ - Kind: interfaces.DataSourceErrorKindInvalidData, - Message: err.Error(), - Time: time.Now(), - }) - return - } - } - merged, err := filedata.Merge(filedata.DuplicateKeysHandling(fs.duplicateKeysHandling), docs...) - if err == nil { - storeData := []ldstoretypes.Collection{ - {Kind: datakinds.Features, Items: merged.Flags}, - {Kind: datakinds.Segments, Items: merged.Segments}, - } - if fs.dataSourceUpdates.Init(storeData) { - fs.signalStartComplete(true) - fs.dataSourceUpdates.UpdateStatus(interfaces.DataSourceStateValid, interfaces.DataSourceErrorInfo{}) - } - } else { - fs.dataSourceUpdates.UpdateStatus(interfaces.DataSourceStateInterrupted, - interfaces.DataSourceErrorInfo{ - Kind: interfaces.DataSourceErrorKindInvalidData, - Message: err.Error(), - Time: time.Now(), - }) +func (fs *fileDataSource) applyData(merged filedata.MergeResult) { + storeData := []ldstoretypes.Collection{ + {Kind: datakinds.Features, Items: merged.Flags}, + {Kind: datakinds.Segments, Items: merged.Segments}, } - if err != nil { - fs.loggers.Error(err) + if fs.dataSourceUpdates.Init(storeData) { + fs.signalStartComplete(true) + fs.dataSourceUpdates.UpdateStatus(interfaces.DataSourceStateValid, interfaces.DataSourceErrorInfo{}) } } +func (fs *fileDataSource) handleError(err error) { + fs.dataSourceUpdates.UpdateStatus(interfaces.DataSourceStateInterrupted, + interfaces.DataSourceErrorInfo{ + Kind: interfaces.DataSourceErrorKindInvalidData, + Message: err.Error(), + Time: time.Now(), + }) +} + func (fs *fileDataSource) signalStartComplete(succeeded bool) { fs.readyOnce.Do(func() { fs.isInitialized = succeeded @@ -134,6 +127,9 @@ func (fs *fileDataSource) Close() (err error) { if fs.closeReloaderCh != nil { close(fs.closeReloaderCh) } + if fs.reloader != nil { + fs.reloader.Close() + } }) return nil } diff --git a/ldfiledatav2/file_data_source_impl.go b/ldfiledatav2/file_data_source_impl.go index 0f988b5e..eca025bc 100644 --- a/ldfiledatav2/file_data_source_impl.go +++ b/ldfiledatav2/file_data_source_impl.go @@ -3,7 +3,6 @@ package ldfiledatav2 import ( "context" "errors" - "fmt" "sync/atomic" "time" @@ -26,6 +25,7 @@ type fileDataSource struct { absFilePaths []string duplicateKeysHandling DuplicateKeysHandling reloaderFactory ReloaderFactory + reloader *filedata.Reloader loggers ldlog.Loggers closeReloaderCh chan struct{} @@ -79,10 +79,28 @@ func (fs *fileDataSource) Sync(ds subsystems.DataSelector) <-chan subsystems.Dat return resultChan } - fs.reload() + // Debouncing and automatic retries only matter when something can trigger further + // reloads; a source configured without a reloader loads exactly once. + var debounceDelay, retryDelay time.Duration + if fs.reloaderFactory != nil { + debounceDelay = filedata.DefaultDebounceDelay + retryDelay = filedata.DefaultRetryDelay + } + fs.reloader = filedata.NewReloader(filedata.ReloaderConfig{ + Paths: fs.absFilePaths, + DuplicateKeysHandling: filedata.DuplicateKeysHandling(fs.duplicateKeysHandling), + Loggers: fs.loggers, + Apply: fs.applyData, + OnError: fs.handleError, + DebounceDelay: debounceDelay, + RetryDelay: retryDelay, + SkipUnchanged: true, + }) + fs.reloader.ReloadNow() + if fs.reloaderFactory != nil { fs.closeReloaderCh = make(chan struct{}) - err := fs.reloaderFactory(fs.absFilePaths, fs.loggers, fs.reload, fs.closeReloaderCh) + err := fs.reloaderFactory(fs.absFilePaths, fs.loggers, fs.reloader.Trigger, fs.closeReloaderCh) if err != nil { fs.loggers.Errorf("Unable to start reloader: %s\n", err) result.State = interfaces.DataSourceStateOff @@ -127,7 +145,19 @@ func (fs *fileDataSource) Sync(ds subsystems.DataSelector) <-chan subsystems.Dat } func (fs *fileDataSource) Fetch(ds subsystems.DataSelector, ctx context.Context) (*subsystems.Basis, bool, error) { - changeSet, err := fs.load() + docs := make([]filedata.Document, 0, len(fs.absFilePaths)) + for _, path := range fs.absFilePaths { + doc, err := filedata.ReadFile(path) + if err != nil { + return nil, false, &filedata.ReadError{Err: err, Path: path} + } + docs = append(docs, doc) + } + merged, err := filedata.Merge(filedata.DuplicateKeysHandling(fs.duplicateKeysHandling), docs...) + if err != nil { + return nil, false, err + } + changeSet, err := fs.makeChangeSet(merged) if err != nil { return nil, false, err } @@ -137,69 +167,35 @@ func (fs *fileDataSource) Fetch(ds subsystems.DataSelector, ctx context.Context) }, false, nil } -// Reload tells the data source to immediately attempt to reread all of the configured source files -// and update the feature flag state. If any file cannot be loaded or parsed, the flag state will not -// be modified. -func (fs *fileDataSource) reload() { - if fs.closeReloaderCh != nil { - fs.loggers.Info("Reloading flag data after detecting a change") - } - - changeSet, err := fs.load() +func (fs *fileDataSource) applyData(merged filedata.MergeResult) { + changeSet, err := fs.makeChangeSet(merged) if err == nil { fs.changeSetBroadcaster.Broadcast(*changeSet) } else { - fs.loggers.Errorf("Unable to load flags: %s", err) - errorKind := interfaces.DataSourceErrorKindInvalidData - var readErr *fileReadError - if errors.As(err, &readErr) { - errorKind = interfaces.DataSourceErrorKindUnknown - } - fs.statusBroadcaster.Broadcast(interfaces.DataSynchronizerStatus{ - State: interfaces.DataSourceStateInterrupted, - Error: interfaces.DataSourceErrorInfo{ - Kind: errorKind, - StatusCode: 0, - Message: err.Error(), - Time: time.Time{}, - }, - FallbackToFDv1: false, - }) + fs.handleError(err) } } -// fileReadError distinguishes a failure to read or parse one of the source files from a -// failure to merge their contents. -type fileReadError struct { - err error - path string -} - -func (e *fileReadError) Error() string { - return fmt.Sprintf("%s [%s]", e.err, e.path) -} - -func (e *fileReadError) Unwrap() error { - return e.err -} - -// load synchronously reads and merges all of the configured source files, returning the -// result as a full-transfer change set. -func (fs *fileDataSource) load() (*subsystems.ChangeSet, error) { - docs := make([]filedata.Document, 0) - for _, path := range fs.absFilePaths { - doc, err := filedata.ReadFile(path) - if err != nil { - return nil, &fileReadError{err: err, path: path} - } - docs = append(docs, doc) - } - - merged, err := filedata.Merge(filedata.DuplicateKeysHandling(fs.duplicateKeysHandling), docs...) - if err != nil { - return nil, err +func (fs *fileDataSource) handleError(err error) { + errorKind := interfaces.DataSourceErrorKindInvalidData + var readErr *filedata.ReadError + if errors.As(err, &readErr) { + errorKind = interfaces.DataSourceErrorKindUnknown } + fs.statusBroadcaster.Broadcast(interfaces.DataSynchronizerStatus{ + State: interfaces.DataSourceStateInterrupted, + Error: interfaces.DataSourceErrorInfo{ + Kind: errorKind, + StatusCode: 0, + Message: err.Error(), + Time: time.Time{}, + }, + FallbackToFDv1: false, + }) +} +// makeChangeSet expresses a merged file data set as a full-transfer change set. +func (fs *fileDataSource) makeChangeSet(merged filedata.MergeResult) (*subsystems.ChangeSet, error) { fs.version++ intent := subsystems.ServerIntent{ Payload: subsystems.Payload{ @@ -239,6 +235,9 @@ func (fs *fileDataSource) Close() (err error) { if fs.closeReloaderCh != nil { close(fs.closeReloaderCh) } + if fs.reloader != nil { + fs.reloader.Close() + } return nil // already closed } diff --git a/ldfiledatav2/file_data_source_test.go b/ldfiledatav2/file_data_source_test.go index a5d42d09..bb30cb92 100644 --- a/ldfiledatav2/file_data_source_test.go +++ b/ldfiledatav2/file_data_source_test.go @@ -12,6 +12,7 @@ import ( "github.com/launchdarkly/go-server-sdk/v7/subsystems" th "github.com/launchdarkly/go-test-helpers/v3" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestSuccessfullyLoadsJsonFlags(t *testing.T) { @@ -234,7 +235,16 @@ func TestRecoversFromInvalidDataWhenFileUpdated(t *testing.T) { reloadCh <- struct{}{} - result = <-resultChan + // The failed initial load schedules automatic retries, so additional interrupted + // results may arrive before the reload that observes the corrected file. + deadline := time.After(5 * time.Second) + for result.State != interfaces.DataSourceStateValid { + select { + case result = <-resultChan: + case <-deadline: + require.FailNow(t, "timed out waiting for a valid result") + } + } assert.NotNil(t, result.ChangeSet) assert.Len(t, result.ChangeSet.Changes(), 1) assert.Equal(t, "my-flag", result.ChangeSet.Changes()[0].Key) From 1c754e7f15d98b24439f4bc4d586ae358c6aef29 Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Wed, 8 Jul 2026 10:17:40 -0700 Subject: [PATCH 3/8] fix: address review findings on file data source lifecycle and versioning Creates the reloader close channel up front so Close never races a concurrent Sync/Start assigning it (an explicit flag now gates the change-driven reload log instead); makes the change-set version counter atomic since loads can run concurrently from Fetch and the reloader; adds regression tests proving Close closes the channel given to the reloader; and documents MergeResult's ordering guarantee precisely (deterministic at document granularity only). --- internal/filedata/merge.go | 10 +++++++--- ldfiledata/file_data_source_impl.go | 16 +++++++++------ ldfiledata/file_data_source_test.go | 27 ++++++++++++++++++++++++++ ldfiledatav2/file_data_source_impl.go | 25 +++++++++++++----------- ldfiledatav2/file_data_source_test.go | 28 +++++++++++++++++++++++++++ 5 files changed, 86 insertions(+), 20 deletions(-) diff --git a/internal/filedata/merge.go b/internal/filedata/merge.go index 302ed225..cd340200 100644 --- a/internal/filedata/merge.go +++ b/internal/filedata/merge.go @@ -21,9 +21,13 @@ const ( DuplicateKeysIgnoreAllButFirst DuplicateKeysHandling = "ignore" ) -// MergeResult holds the merged items from one or more documents. Items appear in the order -// they were first encountered, processing documents in the order they were given. Item -// values are *ldmodel.FeatureFlag or *ldmodel.Segment. +// MergeResult holds the merged items from one or more documents. Item values are +// *ldmodel.FeatureFlag or *ldmodel.Segment. +// +// Ordering is deterministic at document granularity only: all of one document's items +// precede the next document's, matching the order the documents were given, but the +// relative order of items within a single document is unspecified. Consumers key items by +// their Key and must not rely on within-document ordering. type MergeResult struct { Flags []ldstoretypes.KeyedItemDescriptor Segments []ldstoretypes.KeyedItemDescriptor diff --git a/ldfiledata/file_data_source_impl.go b/ldfiledata/file_data_source_impl.go index 1689028b..466471a4 100644 --- a/ldfiledata/file_data_source_impl.go +++ b/ldfiledata/file_data_source_impl.go @@ -22,7 +22,12 @@ type fileDataSource struct { readyCh chan<- struct{} readyOnce sync.Once closeOnce sync.Once - closeReloaderCh chan struct{} + // closeReloaderCh is created up front rather than when the reloader starts, so that + // Close never races with Start assigning it. + closeReloaderCh chan struct{} + // reloaderStarted means reload calls may now come from the reloader, which is worth a + // log line; the initial load is not. + reloaderStarted bool } func newFileDataSourceImpl( @@ -44,6 +49,7 @@ func newFileDataSourceImpl( duplicateKeysHandling: duplicateKeysHandling, reloaderFactory: reloaderFactory, loggers: context.GetLogging().Loggers, + closeReloaderCh: make(chan struct{}), } fs.loggers.SetPrefix("FileDataSource:") return fs, nil @@ -66,7 +72,7 @@ func (fs *fileDataSource) Start(closeWhenReady chan<- struct{}) { // If there is a reloader, and if we haven't yet successfully loaded data, then the // readiness signal will happen the first time we do get valid data (in reload). - fs.closeReloaderCh = make(chan struct{}) + fs.reloaderStarted = true err := fs.reloaderFactory(fs.absFilePaths, fs.loggers, fs.reload, fs.closeReloaderCh) if err != nil { fs.loggers.Errorf("Unable to start reloader: %s\n", err) @@ -77,7 +83,7 @@ func (fs *fileDataSource) Start(closeWhenReady chan<- struct{}) { // and update the feature flag state. If any file cannot be loaded or parsed, the flag state will not // be modified. func (fs *fileDataSource) reload() { - if fs.closeReloaderCh != nil { + if fs.reloaderStarted { fs.loggers.Info("Reloading flag data after detecting a change") } docs := make([]filedata.Document, 0) @@ -131,9 +137,7 @@ func (fs *fileDataSource) signalStartComplete(succeeded bool) { // Close is called automatically when the client is closed. func (fs *fileDataSource) Close() (err error) { fs.closeOnce.Do(func() { - if fs.closeReloaderCh != nil { - close(fs.closeReloaderCh) - } + close(fs.closeReloaderCh) }) return nil } diff --git a/ldfiledata/file_data_source_test.go b/ldfiledata/file_data_source_test.go index 04ca4553..b0553b41 100644 --- a/ldfiledata/file_data_source_test.go +++ b/ldfiledata/file_data_source_test.go @@ -4,6 +4,7 @@ import ( "errors" "os" "testing" + "time" "github.com/launchdarkly/go-server-sdk/v7/internal/sharedtest/mocks" @@ -253,3 +254,29 @@ func requireSegment(t *testing.T, store subsystems.DataStore, key string) *ldmod require.NotNil(t, item.Item) return item.Item.(*ldmodel.Segment) } + +func TestCloseStopsReloader(t *testing.T) { + th.WithTempFileData([]byte(`{"flags": {"my-flag": {"on": true}}}`), func(filename string) { + reloaderCloseCh := make(chan (<-chan struct{}), 1) + f := func(paths []string, loggers ldlog.Loggers, reload func(), closeCh <-chan struct{}) error { + reloaderCloseCh <- closeCh + return nil + } + + factory := DataSource().FilePaths(filename).Reloader(f) + withFileDataSourceTestParams(factory, func(p fileDataSourceTestParams) { + p.waitForStart() + + closeCh := <-reloaderCloseCh + assert.NoError(t, p.dataSource.Close()) + + // The channel given to the reloader must be closed, so that whatever goroutines the + // reloader started can terminate. + select { + case <-closeCh: + case <-time.After(time.Second): + assert.Fail(t, "reloader close channel was not closed by Close()") + } + }) + }) +} diff --git a/ldfiledatav2/file_data_source_impl.go b/ldfiledatav2/file_data_source_impl.go index 0f988b5e..daec3a50 100644 --- a/ldfiledatav2/file_data_source_impl.go +++ b/ldfiledatav2/file_data_source_impl.go @@ -20,14 +20,20 @@ type fileDataSource struct { changeSetBroadcaster *internal.Broadcaster[subsystems.ChangeSet] statusBroadcaster *internal.Broadcaster[interfaces.DataSynchronizerStatus] // NOTE: this is not really used anymore because file data sources at this - // moment will not report a selector. - version int + // moment will not report a selector. It is atomic because loads can happen + // concurrently from Fetch and from the reloader. + version atomic.Int64 absFilePaths []string duplicateKeysHandling DuplicateKeysHandling reloaderFactory ReloaderFactory loggers ldlog.Loggers - closeReloaderCh chan struct{} + // closeReloaderCh is created up front rather than when the reloader starts, so that + // Close never races with Sync assigning it. + closeReloaderCh chan struct{} + // reloaderStarted means reload calls may now come from the reloader, which is worth a + // log line; the initial load is not. + reloaderStarted bool closed atomic.Bool quit chan struct{} @@ -52,6 +58,7 @@ func newFileDataSourceImpl( duplicateKeysHandling: duplicateKeysHandling, reloaderFactory: reloaderFactory, loggers: context.GetLogging().Loggers, + closeReloaderCh: make(chan struct{}), quit: make(chan struct{}), } fs.loggers.SetPrefix("FileDataSource:") @@ -81,7 +88,7 @@ func (fs *fileDataSource) Sync(ds subsystems.DataSelector) <-chan subsystems.Dat fs.reload() if fs.reloaderFactory != nil { - fs.closeReloaderCh = make(chan struct{}) + fs.reloaderStarted = true err := fs.reloaderFactory(fs.absFilePaths, fs.loggers, fs.reload, fs.closeReloaderCh) if err != nil { fs.loggers.Errorf("Unable to start reloader: %s\n", err) @@ -141,7 +148,7 @@ func (fs *fileDataSource) Fetch(ds subsystems.DataSelector, ctx context.Context) // and update the feature flag state. If any file cannot be loaded or parsed, the flag state will not // be modified. func (fs *fileDataSource) reload() { - if fs.closeReloaderCh != nil { + if fs.reloaderStarted { fs.loggers.Info("Reloading flag data after detecting a change") } @@ -200,11 +207,10 @@ func (fs *fileDataSource) load() (*subsystems.ChangeSet, error) { return nil, err } - fs.version++ intent := subsystems.ServerIntent{ Payload: subsystems.Payload{ ID: "", - Target: fs.version, + Target: int(fs.version.Add(1)), Code: subsystems.IntentTransferFull, Reason: "payload-missing", }, @@ -235,10 +241,7 @@ func (fs *fileDataSource) load() (*subsystems.ChangeSet, error) { func (fs *fileDataSource) Close() (err error) { if swapped := fs.closed.CompareAndSwap(false, true); swapped { close(fs.quit) - - if fs.closeReloaderCh != nil { - close(fs.closeReloaderCh) - } + close(fs.closeReloaderCh) return nil // already closed } diff --git a/ldfiledatav2/file_data_source_test.go b/ldfiledatav2/file_data_source_test.go index a5d42d09..210944ed 100644 --- a/ldfiledatav2/file_data_source_test.go +++ b/ldfiledatav2/file_data_source_test.go @@ -315,3 +315,31 @@ func TestInitializerReturnsErrorIfFileDoesNotExist(t *testing.T) { assert.Error(t, err) }) } + +func TestCloseStopsReloader(t *testing.T) { + th.WithTempFileData([]byte(`{"flags": {"my-flag": {"on": true}}}`), func(filename string) { + reloaderCloseCh := make(chan (<-chan struct{}), 1) + f := func(paths []string, loggers ldlog.Loggers, reload func(), closeCh <-chan struct{}) error { + reloaderCloseCh <- closeCh + return nil + } + + factory := DataSource().FilePaths(filename).Reloader(f) + sync, err := factory.Build(subsystems.BasicClientContext{}) + assert.NoError(t, err) + + resultChan := sync.Sync(mocks.NewMockDataSelector(subsystems.NoSelector())) + <-resultChan + + closeCh := <-reloaderCloseCh + assert.NoError(t, sync.Close()) + + // The channel given to the reloader must be closed, so that whatever goroutines the + // reloader started can terminate. + select { + case <-closeCh: + case <-time.After(time.Second): + assert.Fail(t, "reloader close channel was not closed by Close()") + } + }) +} From 41f0f21b149e2f446d74f39d02d2477e930cbd28 Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Wed, 8 Jul 2026 13:26:51 -0700 Subject: [PATCH 4/8] fix: address review findings on reloader shutdown and failure reporting Close no longer waits for an in-flight reload, so a read wedged on unresponsive storage cannot wedge shutdown; a reload instead re-checks the closed state before delivering through Apply/OnError. OnError now fires once per distinct failure rather than on every automatic retry, sparing status listeners a broadcast per second for a permanently broken file. The Reloader is created in the file data source constructors, so a repeated Sync cannot orphan an earlier instance and Close never races an assignment. Also: Apply nil-guarded like OnError; a failed synchronous ReloadNow arms the retry timer directly instead of routing through the change-trigger path (whose log line claimed a change was detected); zero-vs-negative delay semantics documented; the read-error message built in one place; debounce test timing ratio widened. --- internal/filedata/document.go | 6 +- internal/filedata/reloader.go | 105 ++++++++++++++++++-------- internal/filedata/reloader_test.go | 70 ++++++++++++++++- ldfiledata/file_data_source_impl.go | 28 +++---- ldfiledatav2/file_data_source_impl.go | 41 +++++----- 5 files changed, 181 insertions(+), 69 deletions(-) diff --git a/internal/filedata/document.go b/internal/filedata/document.go index 2e250d63..a881590a 100644 --- a/internal/filedata/document.go +++ b/internal/filedata/document.go @@ -44,11 +44,15 @@ func (e *ReadError) Unwrap() error { func ReadFile(path string) (Document, error) { rawData, err := os.ReadFile(path) //nolint:gosec // G304: ok to read file into variable if err != nil { - return Document{}, fmt.Errorf("unable to read file: %s", err) + return Document{}, wrapReadError(err) } return parseDocument(rawData) } +func wrapReadError(err error) error { + return fmt.Errorf("unable to read file: %s", err) +} + func parseDocument(rawData []byte) (Document, error) { var data Document var err error diff --git a/internal/filedata/reloader.go b/internal/filedata/reloader.go index fb52ecc9..e1e2e646 100644 --- a/internal/filedata/reloader.go +++ b/internal/filedata/reloader.go @@ -3,7 +3,6 @@ package filedata import ( "bytes" "crypto/sha256" - "errors" "os" "sync" "time" @@ -33,19 +32,23 @@ type ReloaderConfig struct { Loggers ldlog.Loggers // Apply is invoked with each successfully merged result. Calls are serialized on the // reloader's own goroutine (or, for ReloadNow, under the same lock), so implementations - // do not need their own synchronization against other reloads. + // do not need their own synchronization against other reloads. Apply and OnError must + // not call back into Close. Apply func(MergeResult) - // OnError is invoked for each failed reload. The error is a *ReadError when a file could - // not be read or parsed, or a merge error otherwise. The reloader logs failures itself, - // so implementations only need to update their own state. + // OnError is invoked when a reload fails, once per distinct failure: with automatic + // retries, repeats of an identical failure do not re-invoke it (a success re-arms it). + // The error is a *ReadError when a file could not be read or parsed, or a merge error + // otherwise. The reloader logs failures itself, so implementations only need to update + // their own state. OnError func(err error) // DebounceDelay is how long to wait after a Trigger call for further calls to settle - // before reloading, coalescing bursts of change notifications into one reload. If zero, - // each Trigger reloads immediately. + // before reloading, coalescing bursts of change notifications into one reload. If zero + // or negative, each Trigger reloads immediately. DebounceDelay time.Duration // RetryDelay is how long to wait after a failed reload before automatically retrying, // so that a failure observed while a file was being rewritten recovers even if no - // further change notification arrives. If zero, there is no automatic retry. + // further change notification arrives. If zero or negative, there is no automatic + // retry. RetryDelay time.Duration // SkipUnchanged, if true, suppresses the Apply call when the files' raw contents are // byte-identical to the last successfully applied contents. @@ -58,9 +61,11 @@ type ReloaderConfig struct { type Reloader struct { cfg ReloaderConfig triggerCh chan struct{} - closeCh chan struct{} - doneCh chan struct{} - closeOnce sync.Once + // retryRequestCh asks the run goroutine to arm the retry timer without performing an + // immediate reload; it is used when a synchronous ReloadNow fails. + retryRequestCh chan struct{} + closeCh chan struct{} + closeOnce sync.Once // reloadMu serializes the actual load work between ReloadNow and the run goroutine. reloadMu sync.Mutex @@ -72,10 +77,10 @@ type Reloader struct { // ReloadNow, route change signals to Trigger, and call Close when finished. func NewReloader(cfg ReloaderConfig) *Reloader { r := &Reloader{ - cfg: cfg, - triggerCh: make(chan struct{}, 1), - closeCh: make(chan struct{}), - doneCh: make(chan struct{}), + cfg: cfg, + triggerCh: make(chan struct{}, 1), + retryRequestCh: make(chan struct{}, 1), + closeCh: make(chan struct{}), } go r.run() return r @@ -86,7 +91,10 @@ func NewReloader(cfg ReloaderConfig) *Reloader { // failed triggered reload. func (r *Reloader) ReloadNow() { if !r.reload() && r.cfg.RetryDelay > 0 { - r.Trigger() + select { + case r.retryRequestCh <- struct{}{}: + default: + } } } @@ -100,20 +108,19 @@ func (r *Reloader) Trigger() { } } -// Close stops the reloader. No reload will call Apply or OnError after Close returns. +// Close stops the reloader. It does not wait for a reload that is already in progress — +// one wedged in a blocking file read (an unresponsive network mount, for example) must not +// be able to wedge shutdown — so such a reload may still deliver its result through Apply +// or OnError shortly after Close returns; consumers must tolerate that, as they always +// have for late reloads. A reload that has not yet reached its callbacks when Close is +// called will not invoke them. func (r *Reloader) Close() { r.closeOnce.Do(func() { close(r.closeCh) - <-r.doneCh - // An in-flight reload holds reloadMu, so acquiring it guarantees the reload has - // finished before Close returns. - r.reloadMu.Lock() - defer r.reloadMu.Unlock() }) } func (r *Reloader) run() { - defer close(r.doneCh) var debounceTimer, retryTimer *time.Timer var debounceC, retryC <-chan time.Time stopTimer := func(timer *time.Timer) { @@ -155,9 +162,20 @@ func (r *Reloader) run() { debounceTimer = time.NewTimer(r.cfg.DebounceDelay) debounceC = debounceTimer.C } else { + // Stop-then-Reset without a drain is safe under this module's Go version: + // an expired timer's tick cannot be delivered after Reset. At worst a tick + // already consumed by the debounceC case races a new Trigger, producing one + // extra serialized reload that skip-unchanged absorbs. stopTimer(debounceTimer) debounceTimer.Reset(r.cfg.DebounceDelay) } + case <-r.retryRequestCh: + // A synchronous ReloadNow failed; arm the retry timer without reloading again + // immediately. An already-armed retry keeps its earlier deadline. + if retryTimer == nil { + retryTimer = time.NewTimer(r.cfg.RetryDelay) + retryC = retryTimer.C + } case <-debounceC: debounceTimer, debounceC = nil, nil reloadAndMaybeRetry(false) @@ -177,10 +195,8 @@ func (r *Reloader) reload() bool { // A trigger already queued when Close was called can still reach here; the select in // run() does not prioritize closeCh over triggerCh. - select { - case <-r.closeCh: + if r.isClosed() { return true - default: } docs := make([]Document, 0, len(r.cfg.Paths)) @@ -188,7 +204,7 @@ func (r *Reloader) reload() bool { for _, path := range r.cfg.Paths { rawData, err := os.ReadFile(path) //nolint:gosec // G304: ok to read file into variable if err != nil { - return r.fail(&ReadError{Err: errors.New("unable to read file: " + err.Error()), Path: path}) + return r.fail(&ReadError{Err: wrapReadError(err), Path: path}) } _, _ = hasher.Write(rawData) _, _ = hasher.Write([]byte{0}) @@ -204,27 +220,50 @@ func (r *Reloader) reload() bool { return r.fail(err) } + // Close may have happened while the files were being read; deliver nothing in that case. + if r.isClosed() { + return true + } + r.lastErrorMsg = "" hash := hasher.Sum(nil) if r.cfg.SkipUnchanged && bytes.Equal(hash, r.lastGoodHash) { return true } r.lastGoodHash = hash - r.cfg.Apply(merged) + if r.cfg.Apply != nil { + r.cfg.Apply(merged) + } return true } func (r *Reloader) fail(err error) bool { - // With automatic retries, a persistent failure would repeat the same log entry on every - // attempt, so repeats of an identical failure are demoted to debug level. + // Close may have happened while the files were being read; deliver nothing in that + // case, and report success so no retry is armed. + if r.isClosed() { + return true + } + // With automatic retries, a persistent failure would repeat the same log entry and the + // same callback on every attempt, so repeats of an identical failure are demoted to + // debug level and do not re-invoke OnError. Status-listening consumers therefore see + // one report per distinct failure rather than one per retry. if err.Error() == r.lastErrorMsg { r.cfg.Loggers.Debugf("Unable to load flags: %s", err) - } else { - r.lastErrorMsg = err.Error() - r.cfg.Loggers.Errorf("Unable to load flags: %s", err) + return false } + r.lastErrorMsg = err.Error() + r.cfg.Loggers.Errorf("Unable to load flags: %s", err) if r.cfg.OnError != nil { r.cfg.OnError(err) } return false } + +func (r *Reloader) isClosed() bool { + select { + case <-r.closeCh: + return true + default: + return false + } +} diff --git a/internal/filedata/reloader_test.go b/internal/filedata/reloader_test.go index 0dddfba8..2c2de6ba 100644 --- a/internal/filedata/reloader_test.go +++ b/internal/filedata/reloader_test.go @@ -7,6 +7,7 @@ import ( "time" "github.com/launchdarkly/go-sdk-common/v3/ldlog" + "github.com/launchdarkly/go-sdk-common/v3/ldlogtest" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -101,8 +102,10 @@ func TestReloaderReportsFailureAndRetainsNothing(t *testing.T) { } func TestReloaderDebounceCoalescesTriggers(t *testing.T) { + // The settle window is much longer than the whole trigger burst, so that a scheduling + // stall during the burst cannot let the debounce fire early and split the reloads. f := newReloaderFixture(t, `{"flagValues": {"flag1": true}}`, func(cfg *ReloaderConfig) { - cfg.DebounceDelay = 20 * time.Millisecond + cfg.DebounceDelay = 400 * time.Millisecond }) f.write(t, `{"flagValues": {"flag1": false}}`) for i := 0; i < 20; i++ { @@ -114,6 +117,71 @@ func TestReloaderDebounceCoalescesTriggers(t *testing.T) { f.requireQuiet(t, 100*time.Millisecond) } +func TestReloaderReportsIdenticalFailureOnlyOnce(t *testing.T) { + logCapture := ldlogtest.NewMockLog() + f := newReloaderFixture(t, `{"flagValues"`, func(cfg *ReloaderConfig) { + cfg.RetryDelay = 10 * time.Millisecond + cfg.Loggers = logCapture.Loggers + }) + f.reloader.ReloadNow() + f.requireErrored(t) + + // The automatic retries keep failing identically; the error is neither re-reported to + // OnError nor re-logged at error level. + f.requireQuiet(t, 200*time.Millisecond) + assert.Len(t, logCapture.GetOutput(ldlog.Error), 1) + + // A different failure is a new report. + f.write(t, `{"flagValues": {bad}}`) + err := f.requireErrored(t) + assert.Contains(t, err.Error(), "error parsing file") + + // Success re-arms reporting: the same failure recurring afterward is reported again. + f.write(t, `{"flagValues": {"flag1": true}}`) + f.requireApplied(t) + f.write(t, `{"flagValues"`) + f.reloader.Trigger() + f.requireErrored(t) +} + +func TestReloaderCloseDoesNotWaitForInFlightReload(t *testing.T) { + applyEntered := make(chan struct{}) + applyRelease := make(chan struct{}) + path := filepath.Join(t.TempDir(), "data.json") + require.NoError(t, os.WriteFile(path, []byte(`{"flagValues": {"flag1": true}}`), 0600)) + + r := NewReloader(ReloaderConfig{ + Paths: []string{path}, + DuplicateKeysHandling: DuplicateKeysFail, + Loggers: ldlog.NewDisabledLoggers(), + Apply: func(MergeResult) { + close(applyEntered) + <-applyRelease + }, + }) + defer close(applyRelease) + + r.Trigger() + select { + case <-applyEntered: + case <-time.After(testTimeout): + require.FailNow(t, "timed out waiting for the reload to start") + } + + // The reload is parked inside Apply; Close must return anyway, because a reload wedged + // in blocking I/O must not be able to wedge shutdown. + closed := make(chan struct{}) + go func() { + r.Close() + close(closed) + }() + select { + case <-closed: + case <-time.After(testTimeout): + require.FailNow(t, "Close blocked on an in-flight reload") + } +} + func TestReloaderRetriesAfterFailureWithoutFurtherTriggers(t *testing.T) { f := newReloaderFixture(t, `{"flagValues": {"flag1": true}}`, func(cfg *ReloaderConfig) { cfg.RetryDelay = 20 * time.Millisecond diff --git a/ldfiledata/file_data_source_impl.go b/ldfiledata/file_data_source_impl.go index 3026f7ba..b69f79a5 100644 --- a/ldfiledata/file_data_source_impl.go +++ b/ldfiledata/file_data_source_impl.go @@ -50,20 +50,13 @@ func newFileDataSourceImpl( closeReloaderCh: make(chan struct{}), } fs.loggers.SetPrefix("FileDataSource:") - return fs, nil -} - -func (fs *fileDataSource) IsInitialized() bool { - return fs.isInitialized -} - -func (fs *fileDataSource) Start(closeWhenReady chan<- struct{}) { - fs.readyCh = closeWhenReady // Debouncing and automatic retries only matter when something can trigger further - // reloads; a source configured without a reloader loads exactly once. + // reloads; a source configured without a reloader loads exactly once. Like + // closeReloaderCh, the Reloader is created up front so that Close never races an + // assignment made in Start. var debounceDelay, retryDelay time.Duration - if fs.reloaderFactory != nil { + if reloaderFactory != nil { debounceDelay = filedata.DefaultDebounceDelay retryDelay = filedata.DefaultRetryDelay } @@ -77,6 +70,15 @@ func (fs *fileDataSource) Start(closeWhenReady chan<- struct{}) { RetryDelay: retryDelay, SkipUnchanged: true, }) + return fs, nil +} + +func (fs *fileDataSource) IsInitialized() bool { + return fs.isInitialized +} + +func (fs *fileDataSource) Start(closeWhenReady chan<- struct{}) { + fs.readyCh = closeWhenReady fs.reloader.ReloadNow() // If there is no reloader, then we signal readiness immediately regardless of whether the @@ -127,9 +129,7 @@ func (fs *fileDataSource) signalStartComplete(succeeded bool) { func (fs *fileDataSource) Close() (err error) { fs.closeOnce.Do(func() { close(fs.closeReloaderCh) - if fs.reloader != nil { - fs.reloader.Close() - } + fs.reloader.Close() }) return nil } diff --git a/ldfiledatav2/file_data_source_impl.go b/ldfiledatav2/file_data_source_impl.go index 43d48591..a428d397 100644 --- a/ldfiledatav2/file_data_source_impl.go +++ b/ldfiledatav2/file_data_source_impl.go @@ -59,6 +59,26 @@ func newFileDataSourceImpl( quit: make(chan struct{}), } fs.loggers.SetPrefix("FileDataSource:") + + // Debouncing and automatic retries only matter when something can trigger further + // reloads; a source configured without a reloader loads exactly once. Like + // closeReloaderCh, the Reloader is created up front so that Close never races an + // assignment made in Sync, and a repeated Sync cannot orphan an earlier instance. + var debounceDelay, retryDelay time.Duration + if reloaderFactory != nil { + debounceDelay = filedata.DefaultDebounceDelay + retryDelay = filedata.DefaultRetryDelay + } + fs.reloader = filedata.NewReloader(filedata.ReloaderConfig{ + Paths: fs.absFilePaths, + DuplicateKeysHandling: filedata.DuplicateKeysHandling(fs.duplicateKeysHandling), + Loggers: fs.loggers, + Apply: fs.applyData, + OnError: fs.handleError, + DebounceDelay: debounceDelay, + RetryDelay: retryDelay, + SkipUnchanged: true, + }) return fs, nil } @@ -83,23 +103,6 @@ func (fs *fileDataSource) Sync(ds subsystems.DataSelector) <-chan subsystems.Dat return resultChan } - // Debouncing and automatic retries only matter when something can trigger further - // reloads; a source configured without a reloader loads exactly once. - var debounceDelay, retryDelay time.Duration - if fs.reloaderFactory != nil { - debounceDelay = filedata.DefaultDebounceDelay - retryDelay = filedata.DefaultRetryDelay - } - fs.reloader = filedata.NewReloader(filedata.ReloaderConfig{ - Paths: fs.absFilePaths, - DuplicateKeysHandling: filedata.DuplicateKeysHandling(fs.duplicateKeysHandling), - Loggers: fs.loggers, - Apply: fs.applyData, - OnError: fs.handleError, - DebounceDelay: debounceDelay, - RetryDelay: retryDelay, - SkipUnchanged: true, - }) fs.reloader.ReloadNow() if fs.reloaderFactory != nil { @@ -234,9 +237,7 @@ func (fs *fileDataSource) Close() (err error) { if swapped := fs.closed.CompareAndSwap(false, true); swapped { close(fs.quit) close(fs.closeReloaderCh) - if fs.reloader != nil { - fs.reloader.Close() - } + fs.reloader.Close() return nil // already closed } From a6e16038cee627eb5fa154fb5555293c0bd7b61b Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Wed, 8 Jul 2026 13:37:44 -0700 Subject: [PATCH 5/8] fix: apply the first successful reload after a failure even when content is unchanged With SkipUnchanged, a success whose bytes matched the last good hash returned without calling Apply. If a failure had intervened, the consumer had heard OnError and moved to an interrupted state, and the skipped Apply meant it never heard about the recovery: both file data sources would report Interrupted indefinitely after a transient read failure whose retry found byte-identical content. The first success after a failure now always applies. --- internal/filedata/reloader.go | 6 +++++- internal/filedata/reloader_test.go | 23 +++++++++++++++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/internal/filedata/reloader.go b/internal/filedata/reloader.go index e1e2e646..5446e212 100644 --- a/internal/filedata/reloader.go +++ b/internal/filedata/reloader.go @@ -225,9 +225,13 @@ func (r *Reloader) reload() bool { return true } + // A success right after a failure must apply even when the content is unchanged since + // the last success: the consumer heard about the failure through OnError and may have + // moved to an interrupted state, and only Apply tells it things are good again. + recovering := r.lastErrorMsg != "" r.lastErrorMsg = "" hash := hasher.Sum(nil) - if r.cfg.SkipUnchanged && bytes.Equal(hash, r.lastGoodHash) { + if r.cfg.SkipUnchanged && !recovering && bytes.Equal(hash, r.lastGoodHash) { return true } r.lastGoodHash = hash diff --git a/internal/filedata/reloader_test.go b/internal/filedata/reloader_test.go index 2c2de6ba..96e1c39d 100644 --- a/internal/filedata/reloader_test.go +++ b/internal/filedata/reloader_test.go @@ -231,6 +231,29 @@ func TestReloaderSkipUnchanged(t *testing.T) { f.requireApplied(t) } +func TestReloaderRecoveryAppliesEvenWhenContentUnchanged(t *testing.T) { + f := newReloaderFixture(t, `{"flagValues": {"flag1": true}}`, func(cfg *ReloaderConfig) { + cfg.SkipUnchanged = true + }) + f.reloader.ReloadNow() + f.requireApplied(t) + + // A reload fails; consumers hear OnError and may move to an interrupted state. + require.NoError(t, os.Remove(f.path)) + f.reloader.Trigger() + f.requireErrored(t) + + // The files come back with byte-identical content. The success must be applied despite + // SkipUnchanged, because only Apply tells the consumer the interruption is over. + f.write(t, `{"flagValues": {"flag1": true}}`) + f.reloader.Trigger() + f.requireApplied(t) + + // Once recovered, identical content skips again. + f.reloader.Trigger() + f.requireQuiet(t, 100*time.Millisecond) +} + func TestReloaderAppliesEveryReloadWhenSkipUnchangedIsOff(t *testing.T) { f := newReloaderFixture(t, `{"flagValues": {"flag1": true}}`, nil) f.reloader.ReloadNow() From 54aeb920b560b8a0f8eb17d7298aecf605205a2e Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Wed, 8 Jul 2026 14:00:13 -0700 Subject: [PATCH 6/8] fix: do not start the reload worker goroutine until the reloader is used Creating the Reloader in the data source constructors meant its worker goroutine started at build time, and a source built as a one-shot initializer -- an interface with no Close that the data system never tears down -- leaked that goroutine for the life of the process, once per build. The worker now starts on the first ReloadNow or Trigger, which the initializer path never calls. Also guards the v2 synchronizer against a repeated Sync (which would accumulate watchers and listeners) with a buffered rejection so the guard itself cannot block, and notes that the safety of a post-Close status broadcast rests on the Broadcaster's lock discipline. --- internal/filedata/reloader.go | 23 +++++++++++++++--- internal/filedata/reloader_test.go | 35 ++++++++++++++++++++++++++- ldfiledata/file_data_source_impl.go | 5 ++++ ldfiledatav2/file_data_source_impl.go | 23 +++++++++++------- ldfiledatav2/file_data_source_test.go | 19 +++++++++++++++ 5 files changed, 91 insertions(+), 14 deletions(-) diff --git a/internal/filedata/reloader.go b/internal/filedata/reloader.go index 5446e212..d0b56592 100644 --- a/internal/filedata/reloader.go +++ b/internal/filedata/reloader.go @@ -66,6 +66,7 @@ type Reloader struct { retryRequestCh chan struct{} closeCh chan struct{} closeOnce sync.Once + startOnce sync.Once // reloadMu serializes the actual load work between ReloadNow and the run goroutine. reloadMu sync.Mutex @@ -73,23 +74,36 @@ type Reloader struct { lastErrorMsg string } -// NewReloader creates a started Reloader. The caller should perform the initial load with +// NewReloader creates a Reloader. The caller should perform the initial load with // ReloadNow, route change signals to Trigger, and call Close when finished. +// +// The worker goroutine starts on the first ReloadNow or Trigger call rather than here: +// a Reloader can be constructed by a component whose lifecycle never uses it (a file data +// source built as a one-shot initializer, whose interface has no Close), and construction +// alone must not leak a goroutine. func NewReloader(cfg ReloaderConfig) *Reloader { - r := &Reloader{ + return &Reloader{ cfg: cfg, triggerCh: make(chan struct{}, 1), retryRequestCh: make(chan struct{}, 1), closeCh: make(chan struct{}), } - go r.run() - return r +} + +func (r *Reloader) ensureStarted() { + if r.isClosed() { + return + } + r.startOnce.Do(func() { + go r.run() + }) } // ReloadNow synchronously loads the files and applies the result (or reports the failure). // It is used for the initial load; a failure here schedules the same automatic retry as a // failed triggered reload. func (r *Reloader) ReloadNow() { + r.ensureStarted() if !r.reload() && r.cfg.RetryDelay > 0 { select { case r.retryRequestCh <- struct{}{}: @@ -102,6 +116,7 @@ func (r *Reloader) ReloadNow() { // debounce delay. It never blocks; signals arriving while a reload is already pending are // coalesced. func (r *Reloader) Trigger() { + r.ensureStarted() select { case r.triggerCh <- struct{}{}: default: diff --git a/internal/filedata/reloader_test.go b/internal/filedata/reloader_test.go index 96e1c39d..cc239150 100644 --- a/internal/filedata/reloader_test.go +++ b/internal/filedata/reloader_test.go @@ -3,6 +3,7 @@ package filedata import ( "os" "path/filepath" + "runtime" "testing" "time" @@ -144,11 +145,32 @@ func TestReloaderReportsIdenticalFailureOnlyOnce(t *testing.T) { f.requireErrored(t) } +func TestReloaderUnusedSpawnsNoGoroutine(t *testing.T) { + // A Reloader can be constructed by a component whose lifecycle never uses or closes it, + // such as a file data source built as a one-shot initializer, so construction alone + // must not start the worker goroutine. + runtime.GC() + before := runtime.NumGoroutine() + path := filepath.Join(t.TempDir(), "data.json") + require.NoError(t, os.WriteFile(path, []byte(`{}`), 0600)) + for range 50 { + NewReloader(ReloaderConfig{ + Paths: []string{path}, + DuplicateKeysHandling: DuplicateKeysFail, + Loggers: ldlog.NewDisabledLoggers(), + Apply: func(MergeResult) {}, + }) + } + assert.LessOrEqual(t, runtime.NumGoroutine(), before+2, + "constructing unused Reloaders must not accumulate goroutines") +} + func TestReloaderCloseDoesNotWaitForInFlightReload(t *testing.T) { applyEntered := make(chan struct{}) applyRelease := make(chan struct{}) path := filepath.Join(t.TempDir(), "data.json") require.NoError(t, os.WriteFile(path, []byte(`{"flagValues": {"flag1": true}}`), 0600)) + goroutinesBefore := runtime.NumGoroutine() r := NewReloader(ReloaderConfig{ Paths: []string{path}, @@ -159,7 +181,6 @@ func TestReloaderCloseDoesNotWaitForInFlightReload(t *testing.T) { <-applyRelease }, }) - defer close(applyRelease) r.Trigger() select { @@ -180,6 +201,18 @@ func TestReloaderCloseDoesNotWaitForInFlightReload(t *testing.T) { case <-time.After(testTimeout): require.FailNow(t, "Close blocked on an in-flight reload") } + + // Once the parked reload is released, the worker goroutine terminates rather than + // lingering past Close. Polled inline: spawning goroutines to poll would disturb the + // count being measured. + close(applyRelease) + deadline := time.Now().Add(testTimeout) + for runtime.NumGoroutine() > goroutinesBefore+1 { + if time.Now().After(deadline) { + require.FailNow(t, "worker goroutine did not exit after Close") + } + time.Sleep(10 * time.Millisecond) + } } func TestReloaderRetriesAfterFailureWithoutFurtherTriggers(t *testing.T) { diff --git a/ldfiledata/file_data_source_impl.go b/ldfiledata/file_data_source_impl.go index b69f79a5..1966fec8 100644 --- a/ldfiledata/file_data_source_impl.go +++ b/ldfiledata/file_data_source_impl.go @@ -126,6 +126,11 @@ func (fs *fileDataSource) signalStartComplete(succeeded bool) { } // Close is called automatically when the client is closed. +// +// Close does not wait for an in-flight reload, so a late applyData/handleError can still +// reach dataSourceUpdates after Close returns. That is safe today because the SDK's status +// broadcaster serializes Broadcast against its own Close with a lock; if that Broadcaster +// discipline ever changes, this ordering needs revisiting. func (fs *fileDataSource) Close() (err error) { fs.closeOnce.Do(func() { close(fs.closeReloaderCh) diff --git a/ldfiledatav2/file_data_source_impl.go b/ldfiledatav2/file_data_source_impl.go index a428d397..63416dec 100644 --- a/ldfiledatav2/file_data_source_impl.go +++ b/ldfiledatav2/file_data_source_impl.go @@ -32,8 +32,9 @@ type fileDataSource struct { // Close never races with Sync assigning it. closeReloaderCh chan struct{} - closed atomic.Bool - quit chan struct{} + closed atomic.Bool + syncStarted atomic.Bool + quit chan struct{} } func newFileDataSourceImpl( @@ -87,6 +88,17 @@ func (fs *fileDataSource) Name() string { } func (fs *fileDataSource) Sync(ds subsystems.DataSelector) <-chan subsystems.DataSynchronizerResult { + // Sync starts the reloader and the file watcher and registers listeners for this + // call's result loop, so it can run at most once: a repeat would accumulate another + // watcher and listener pair feeding nothing. The rejection channel is buffered so an + // unread rejection cannot block. + if fs.closed.Load() || !fs.syncStarted.CompareAndSwap(false, true) { + rejected := make(chan subsystems.DataSynchronizerResult, 1) + rejected <- subsystems.DataSynchronizerResult{State: interfaces.DataSourceStateOff} + close(rejected) + return rejected + } + resultChan := make(chan subsystems.DataSynchronizerResult) changeSetChan := fs.changeSetBroadcaster.AddListener() @@ -96,13 +108,6 @@ func (fs *fileDataSource) Sync(ds subsystems.DataSelector) <-chan subsystems.Dat State: interfaces.DataSourceStateInitializing, } - if fs.closed.Load() { - result.State = interfaces.DataSourceStateOff - resultChan <- result - close(resultChan) - return resultChan - } - fs.reloader.ReloadNow() if fs.reloaderFactory != nil { diff --git a/ldfiledatav2/file_data_source_test.go b/ldfiledatav2/file_data_source_test.go index cc3accca..69005582 100644 --- a/ldfiledatav2/file_data_source_test.go +++ b/ldfiledatav2/file_data_source_test.go @@ -3,6 +3,7 @@ package ldfiledatav2 import ( "context" "os" + "runtime" "testing" "time" @@ -353,3 +354,21 @@ func TestCloseStopsReloader(t *testing.T) { } }) } + +func TestInitializerBuildAndFetchLeaksNoGoroutines(t *testing.T) { + th.WithTempFileData([]byte(`{"flags": {"my-flag": {"on": true}}}`), func(filename string) { + goroutinesBefore := runtime.NumGoroutine() + + // A DataInitializer has no Close and the data system never tears one down, so + // building and fetching must not start anything that outlives the calls. + for i := 0; i < 20; i++ { + initializer, err := DataSource().FilePaths(filename).AsInitializer().Build(subsystems.BasicClientContext{}) + assert.NoError(t, err) + _, _, err = initializer.Fetch(mocks.NewMockDataSelector(subsystems.NoSelector()), context.Background()) + assert.NoError(t, err) + } + + assert.LessOrEqual(t, runtime.NumGoroutine(), goroutinesBefore+2, + "initializer builds must not accumulate goroutines") + }) +} From 8893141edae17424517333b37fc0ab0ab7ee8651 Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Wed, 8 Jul 2026 15:29:06 -0700 Subject: [PATCH 7/8] fix: reject a failed reloader start on a buffered channel instead of deadlocking Sync When the reloader factory failed (for example fsnotify hitting inotify instance or watch limits), the v2 synchronizer sent the Off result on the unbuffered result channel before the reader goroutine existed, so Sync never returned: client startup hung for its full timeout with no fallback attempted, and the already-started reload worker leaked. The failure branch now uses the same buffered single-result rejection as the repeated-Sync guard, stops the reload worker, and detaches the listeners it registered. Also adds the missing tests for this branch and the repeated-Sync rejection, and makes two timing-sensitive tests robust (the goroutine-count check now polls for the process-global count to settle, and the debounce test enables SkipUnchanged like production configs to absorb a legitimate tick-vs-trigger race). --- internal/filedata/reloader_test.go | 15 ++++++-- ldfiledatav2/file_data_source_impl.go | 15 ++++++-- ldfiledatav2/file_data_source_test.go | 54 +++++++++++++++++++++++++++ 3 files changed, 77 insertions(+), 7 deletions(-) diff --git a/internal/filedata/reloader_test.go b/internal/filedata/reloader_test.go index cc239150..276cc0a1 100644 --- a/internal/filedata/reloader_test.go +++ b/internal/filedata/reloader_test.go @@ -105,8 +105,11 @@ func TestReloaderReportsFailureAndRetainsNothing(t *testing.T) { func TestReloaderDebounceCoalescesTriggers(t *testing.T) { // The settle window is much longer than the whole trigger burst, so that a scheduling // stall during the burst cannot let the debounce fire early and split the reloads. + // SkipUnchanged matches production configs and absorbs the one extra serialized reload + // that a debounce tick racing a fresh trigger can legitimately produce. f := newReloaderFixture(t, `{"flagValues": {"flag1": true}}`, func(cfg *ReloaderConfig) { cfg.DebounceDelay = 400 * time.Millisecond + cfg.SkipUnchanged = true }) f.write(t, `{"flagValues": {"flag1": false}}`) for i := 0; i < 20; i++ { @@ -149,7 +152,6 @@ func TestReloaderUnusedSpawnsNoGoroutine(t *testing.T) { // A Reloader can be constructed by a component whose lifecycle never uses or closes it, // such as a file data source built as a one-shot initializer, so construction alone // must not start the worker goroutine. - runtime.GC() before := runtime.NumGoroutine() path := filepath.Join(t.TempDir(), "data.json") require.NoError(t, os.WriteFile(path, []byte(`{}`), 0600)) @@ -161,8 +163,15 @@ func TestReloaderUnusedSpawnsNoGoroutine(t *testing.T) { Apply: func(MergeResult) {}, }) } - assert.LessOrEqual(t, runtime.NumGoroutine(), before+2, - "constructing unused Reloaders must not accumulate goroutines") + // The count is process-global, so goroutines from earlier tests may still be winding + // down; poll until it settles rather than asserting a single sample. + deadline := time.Now().Add(testTimeout) + for runtime.NumGoroutine() > before+2 { + if time.Now().After(deadline) { + require.FailNow(t, "constructing unused Reloaders must not accumulate goroutines") + } + time.Sleep(10 * time.Millisecond) + } } func TestReloaderCloseDoesNotWaitForInFlightReload(t *testing.T) { diff --git a/ldfiledatav2/file_data_source_impl.go b/ldfiledatav2/file_data_source_impl.go index 63416dec..c4d1d70b 100644 --- a/ldfiledatav2/file_data_source_impl.go +++ b/ldfiledatav2/file_data_source_impl.go @@ -114,10 +114,17 @@ func (fs *fileDataSource) Sync(ds subsystems.DataSelector) <-chan subsystems.Dat err := fs.reloaderFactory(fs.absFilePaths, fs.loggers, fs.reloader.Trigger, fs.closeReloaderCh) if err != nil { fs.loggers.Errorf("Unable to start reloader: %s\n", err) - result.State = interfaces.DataSourceStateOff - resultChan <- result - close(resultChan) - return resultChan + // With no reloader there will never be another reload; stop the worker and + // detach this call's listeners so nothing accumulates behind them. The result + // must go on a buffered channel: the reader goroutine below was never started, + // so a send on resultChan would block forever. + fs.reloader.Close() + fs.changeSetBroadcaster.RemoveListener(changeSetChan) + fs.statusBroadcaster.RemoveListener(statusChan) + rejected := make(chan subsystems.DataSynchronizerResult, 1) + rejected <- subsystems.DataSynchronizerResult{State: interfaces.DataSourceStateOff} + close(rejected) + return rejected } } diff --git a/ldfiledatav2/file_data_source_test.go b/ldfiledatav2/file_data_source_test.go index 69005582..068a6d3f 100644 --- a/ldfiledatav2/file_data_source_test.go +++ b/ldfiledatav2/file_data_source_test.go @@ -2,6 +2,7 @@ package ldfiledatav2 import ( "context" + "errors" "os" "runtime" "testing" @@ -372,3 +373,56 @@ func TestInitializerBuildAndFetchLeaksNoGoroutines(t *testing.T) { "initializer builds must not accumulate goroutines") }) } + +func TestSyncReturnsOffWhenReloaderFactoryFails(t *testing.T) { + th.WithTempFileData([]byte(`{"flags": {"my-flag": {"on": true}}}`), func(filename string) { + f := func(paths []string, loggers ldlog.Loggers, reload func(), closeCh <-chan struct{}) error { + return errors.New("no watches available") + } + + factory := DataSource().FilePaths(filename).Reloader(f) + sync, err := factory.Build(subsystems.BasicClientContext{}) + assert.NoError(t, err) + defer sync.Close() + + // The failure must surface as an Off result rather than a hang. + results := make(chan subsystems.DataSynchronizerResult, 1) + go func() { + resultChan := sync.Sync(mocks.NewMockDataSelector(subsystems.NoSelector())) + results <- <-resultChan + _, stillOpen := <-resultChan + assert.False(t, stillOpen, "result channel should be closed after the rejection") + }() + select { + case result := <-results: + assert.Equal(t, interfaces.DataSourceStateOff, result.State) + case <-time.After(5 * time.Second): + require.FailNow(t, "Sync hung after the reloader factory failed") + } + }) +} + +func TestSecondSyncIsRejected(t *testing.T) { + th.WithTempFileData([]byte(`{"flags": {"my-flag": {"on": true}}}`), func(filename string) { + factory := DataSource().FilePaths(filename) + sync, err := factory.Build(subsystems.BasicClientContext{}) + assert.NoError(t, err) + defer sync.Close() + + resultChan := sync.Sync(mocks.NewMockDataSelector(subsystems.NoSelector())) + result := <-resultChan + assert.Equal(t, interfaces.DataSourceStateValid, result.State) + + // Sync is single-use: a repeat is rejected with a single Off result and a closed + // channel, without disturbing the first Sync. + secondChan := sync.Sync(mocks.NewMockDataSelector(subsystems.NoSelector())) + select { + case second := <-secondChan: + assert.Equal(t, interfaces.DataSourceStateOff, second.State) + case <-time.After(5 * time.Second): + require.FailNow(t, "second Sync produced no result") + } + _, stillOpen := <-secondChan + assert.False(t, stillOpen, "second Sync's channel should be closed") + }) +} From 8ca79b2a058fef2716589ed9dae6605af4e6fc8f Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Wed, 8 Jul 2026 15:54:36 -0700 Subject: [PATCH 8/8] test: restore the debounce test's ability to detect broken coalescing Enabling SkipUnchanged had made the test vacuous: with identical file contents, even one reload per trigger collapses to a single Apply. SkipUnchanged is off again and the test now counts applies across the settle window, tolerating the one documented extra reload from a debounce tick racing a fresh trigger and failing on anything more. Verified to fail with debouncing disabled. --- internal/filedata/reloader_test.go | 22 +++++++++++++++++----- 1 file changed, 17 insertions(+), 5 deletions(-) diff --git a/internal/filedata/reloader_test.go b/internal/filedata/reloader_test.go index 276cc0a1..13325c76 100644 --- a/internal/filedata/reloader_test.go +++ b/internal/filedata/reloader_test.go @@ -105,11 +105,11 @@ func TestReloaderReportsFailureAndRetainsNothing(t *testing.T) { func TestReloaderDebounceCoalescesTriggers(t *testing.T) { // The settle window is much longer than the whole trigger burst, so that a scheduling // stall during the burst cannot let the debounce fire early and split the reloads. - // SkipUnchanged matches production configs and absorbs the one extra serialized reload - // that a debounce tick racing a fresh trigger can legitimately produce. + // SkipUnchanged stays off so that every reload is observable as an Apply: with it on, + // identical file contents would collapse even a completely broken debounce into a + // single Apply and the test would prove nothing. f := newReloaderFixture(t, `{"flagValues": {"flag1": true}}`, func(cfg *ReloaderConfig) { cfg.DebounceDelay = 400 * time.Millisecond - cfg.SkipUnchanged = true }) f.write(t, `{"flagValues": {"flag1": false}}`) for i := 0; i < 20; i++ { @@ -117,8 +117,20 @@ func TestReloaderDebounceCoalescesTriggers(t *testing.T) { time.Sleep(time.Millisecond) } f.requireApplied(t) - // All of the triggers, arriving within the settle window, coalesced into one reload. - f.requireQuiet(t, 100*time.Millisecond) + // The burst must coalesce into one reload, with a tolerance of one more: a debounce + // tick racing a fresh trigger can legitimately produce a single extra serialized + // reload, but anything beyond that means triggers are not being coalesced. + extraApplies := 0 + deadline := time.After(600 * time.Millisecond) + for done := false; !done; { + select { + case <-f.applied: + extraApplies++ + case <-deadline: + done = true + } + } + assert.LessOrEqual(t, extraApplies, 1, "trigger burst was not coalesced") } func TestReloaderReportsIdenticalFailureOnlyOnce(t *testing.T) {