chore: Support offline mode #561
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,204 @@ | ||
| package filedata | ||
|
|
||
| import ( | ||
| "context" | ||
| "sync" | ||
| "sync/atomic" | ||
|
|
||
| "github.com/launchdarkly/go-server-sdk/v7/interfaces" | ||
| "github.com/launchdarkly/go-server-sdk/v7/subsystems" | ||
| "github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoretypes" | ||
| ) | ||
|
|
||
| // simpleBroadcaster is a simple implementation of a broadcaster for changeSets and status updates. | ||
| type simpleBroadcaster[T any] struct { | ||
| mu sync.RWMutex | ||
| listeners []chan T | ||
| } | ||
|
|
||
| func newSimpleBroadcaster[T any]() *simpleBroadcaster[T] { | ||
| return &simpleBroadcaster[T]{ | ||
| listeners: make([]chan T, 0), | ||
| } | ||
| } | ||
|
|
||
| func (b *simpleBroadcaster[T]) AddListener() <-chan T { | ||
| b.mu.Lock() | ||
| defer b.mu.Unlock() | ||
| ch := make(chan T, 10) | ||
| b.listeners = append(b.listeners, ch) | ||
| return ch | ||
| } | ||
|
|
||
| func (b *simpleBroadcaster[T]) Broadcast(value T) { | ||
| b.mu.RLock() | ||
| defer b.mu.RUnlock() | ||
| for _, ch := range b.listeners { | ||
| select { | ||
| case ch <- value: | ||
| default: | ||
| // If channel is full, skip this listener | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // OfflineModeSynchronizerFactory creates a synchronizer that loads data from an external source | ||
| // (the relay's archive file) without making any network connections. | ||
| type OfflineModeSynchronizerFactory struct { | ||
| Synchronizer *OfflineModeSynchronizer | ||
| } | ||
|
|
||
| // OfflineModeSynchronizer implements subsystems.DataSynchronizer for offline mode. | ||
| // It loads pre-populated data from the archive file without connecting to LaunchDarkly. | ||
| type OfflineModeSynchronizer struct { | ||
| dataCh <-chan []ldstoretypes.Collection | ||
| changeSetBroadcaster *simpleBroadcaster[subsystems.ChangeSet] | ||
| statusBroadcaster *simpleBroadcaster[interfaces.DataSynchronizerStatus] | ||
| version int32 | ||
| quit chan struct{} | ||
| closed atomic.Bool | ||
| } | ||
|
|
||
| func NewOfflineModeSynchronizer(dataCh <-chan []ldstoretypes.Collection) *OfflineModeSynchronizer { | ||
| return &OfflineModeSynchronizer{ | ||
| dataCh: dataCh, | ||
| changeSetBroadcaster: newSimpleBroadcaster[subsystems.ChangeSet](), | ||
| statusBroadcaster: newSimpleBroadcaster[interfaces.DataSynchronizerStatus](), | ||
| quit: make(chan struct{}), | ||
| } | ||
| } | ||
|
|
||
| func (f OfflineModeSynchronizerFactory) Build( | ||
| ctx subsystems.ClientContext, | ||
| ) (subsystems.DataSynchronizer, error) { | ||
| return f.Synchronizer, nil | ||
| } | ||
|
|
||
| func (s *OfflineModeSynchronizer) Close() error { | ||
| if s.closed.Swap(true) { | ||
| return nil | ||
| } | ||
| close(s.quit) | ||
| return nil | ||
| } | ||
|
|
||
| func (s *OfflineModeSynchronizer) Name() string { | ||
| return "OfflineModeSynchronizer" | ||
| } | ||
|
|
||
| // Fetch returns the current basis (full dataset) from the offline data source. | ||
| func (s *OfflineModeSynchronizer) Fetch(ds subsystems.DataSelector, ctx context.Context) (*subsystems.Basis, error) { | ||
| // Wait for data to arrive from the archive | ||
| select { | ||
| case data := <-s.dataCh: | ||
| changeSet, err := s.makeChangeSetFromCollections(data) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return &subsystems.Basis{ | ||
| ChangeSet: *changeSet, | ||
| Persist: false, | ||
| }, nil | ||
| case <-ctx.Done(): | ||
| return nil, ctx.Err() | ||
| case <-s.quit: | ||
| return nil, context.Canceled | ||
| } | ||
| } | ||
|
|
||
| // Sync starts the synchronizer and returns a channel for receiving updates. | ||
| // For offline mode, this immediately provides the data from the archive file. | ||
| func (s *OfflineModeSynchronizer) Sync(ds subsystems.DataSelector) <-chan subsystems.DataSynchronizerResult { | ||
| resultChan := make(chan subsystems.DataSynchronizerResult) | ||
| changeSetChan := s.changeSetBroadcaster.AddListener() | ||
| statusChan := s.statusBroadcaster.AddListener() | ||
|
|
||
| go func() { | ||
| defer close(resultChan) | ||
|
|
||
| result := subsystems.DataSynchronizerResult{ | ||
| State: interfaces.DataSourceStateInitializing, | ||
| } | ||
|
|
||
| // Wait for initial data from archive | ||
| select { | ||
| case data := <-s.dataCh: | ||
| changeSet, err := s.makeChangeSetFromCollections(data) | ||
| if err != nil { | ||
| result.State = interfaces.DataSourceStateOff | ||
| result.Error = interfaces.DataSourceErrorInfo{ | ||
| Kind: interfaces.DataSourceErrorKindUnknown, | ||
| Message: err.Error(), | ||
| } | ||
| } else { | ||
| result.State = interfaces.DataSourceStateValid | ||
| result.ChangeSet = changeSet | ||
| } | ||
| resultChan <- result | ||
|
|
||
| case <-s.quit: | ||
| result.State = interfaces.DataSourceStateOff | ||
| resultChan <- result | ||
| return | ||
| } | ||
|
|
||
| // Listen for updates (in offline mode, these would be file changes) | ||
| for { | ||
| select { | ||
| case <-s.quit: | ||
| return | ||
| case changeSet, ok := <-changeSetChan: | ||
| if !ok { | ||
| return | ||
| } | ||
| result.ChangeSet = &changeSet | ||
| result.State = interfaces.DataSourceStateValid | ||
| resultChan <- result | ||
|
There was a problem hiding this comment. Bug: Error state persists incorrectly across result updatesThe |
||
| case statusChange, ok := <-statusChan: | ||
| if !ok { | ||
| return | ||
| } | ||
| if statusChange.State != interfaces.DataSourceStateValid { | ||
| result.ChangeSet = nil | ||
| } | ||
| result.State = statusChange.State | ||
| result.Error = statusChange.Error | ||
| resultChan <- result | ||
| } | ||
| } | ||
| }() | ||
|
|
||
| return resultChan | ||
| } | ||
|
|
||
| // makeChangeSetFromCollections converts old-style Collection data to a new-style ChangeSet. | ||
| // This uses the SDK's NewChangeSetFromCollections which pre-caches the collections, | ||
| // avoiding redundant conversions when the data is accessed later. | ||
| func (s *OfflineModeSynchronizer) makeChangeSetFromCollections( | ||
| collections []ldstoretypes.Collection, | ||
| ) (*subsystems.ChangeSet, error) { | ||
| version := int(atomic.AddInt32(&s.version, 1)) | ||
|
|
||
| return subsystems.NewChangeSetFromCollections( | ||
| subsystems.ServerIntent{ | ||
| Payload: subsystems.Payload{ | ||
| ID: "", | ||
| Target: version, | ||
| Code: subsystems.IntentTransferFull, | ||
| Reason: "offline-mode-init", | ||
| }, | ||
| }, | ||
| subsystems.NewSelector("offline", version), | ||
| collections, | ||
| ) | ||
| } | ||
|
|
||
| // UpdateData allows external updates to the data (e.g., when the archive file changes). | ||
| func (s *OfflineModeSynchronizer) UpdateData(collections []ldstoretypes.Collection) error { | ||
| changeSet, err := s.makeChangeSetFromCollections(collections) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| s.changeSetBroadcaster.Broadcast(*changeSet) | ||
| return nil | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Bug: Single-use data channel shared across synchronizer calls
The
dataChchannel is read from by both theFetch()andSync()methods, but only a single data item is ever sent to this channel (inAddEnvironment). Additionally, the sameOfflineModeSynchronizerinstance is used for both primary and fallback synchronizers viaSynchronizers(syncFactory, syncFactory). If the SDK callsSync()on both (or calls bothFetch()andSync()), multiple goroutines will race to receive fromdataCh. Only one will succeed; the others will block forever waiting for data that never arrives, potentially causing initialization hangs.Additional Locations (2)
internal/filedata/offline_mode_synchronizer.go#L110-L143relay/filedata_actions.go#L46-L47