diff --git a/bulker/bulkerapp/app/http_configuration_source.go b/bulker/bulkerapp/app/http_configuration_source.go index 08f7ed3e0..0ffb668f5 100644 --- a/bulker/bulkerapp/app/http_configuration_source.go +++ b/bulker/bulkerapp/app/http_configuration_source.go @@ -83,7 +83,7 @@ type HTTPConfigurationSource struct { } func NewHTTPConfigurationSource(appconfig *Config) *HTTPConfigurationSource { - rep := appbase.NewHTTPRepository[Destinations]("bulker-connections", appconfig.ConfigSource, appconfig.ConfigSourceHTTPAuthToken, appbase.HTTPTagLastModified, &DestinationsRepositoryData{}, 1, appconfig.ConfigRefreshPeriodSec, appconfig.CacheDir) + rep := appbase.NewHTTPRepository[Destinations]("bulker-connections", appconfig.ConfigSource, appconfig.ConfigSourceHTTPAuthToken, appbase.HTTPTagLastModified, &DestinationsRepositoryData{}, 1, appconfig.ConfigRefreshPeriodSec, appconfig.CacheDir, appbase.ExitOnNoData) return &HTTPConfigurationSource{rep} } diff --git a/bulker/config-keeper/app.go b/bulker/config-keeper/app.go index 8097af79e..09df6fd4e 100644 --- a/bulker/config-keeper/app.go +++ b/bulker/config-keeper/app.go @@ -8,6 +8,7 @@ import ( "io" "net/http" "strings" + "sync" "sync/atomic" "time" ) @@ -16,7 +17,58 @@ type Context struct { config *Config server *http.Server pScript appbase.Repository[[]byte] - repositories map[string]appbase.Repository[[]byte] + repositories *repositories +} + +// repositories is a map that is safe for concurrent use. RepositoryHandler adds +// lazily discovered repositories from request goroutines while /health ranges +// over them, and Go aborts the process on a concurrent map read and write. +type repositories struct { + mu sync.RWMutex + byName map[string]appbase.Repository[[]byte] +} + +func newRepositories() *repositories { + return &repositories{byName: map[string]appbase.Repository[[]byte]{}} +} + +func (r *repositories) add(name string, rep appbase.Repository[[]byte]) { + r.mu.Lock() + defer r.mu.Unlock() + r.byName[name] = rep +} + +func (r *repositories) get(name string) (appbase.Repository[[]byte], bool) { + r.mu.RLock() + defer r.mu.RUnlock() + rep, ok := r.byName[name] + return rep, ok +} + +// addIfAbsent returns the repository registered under name afterwards, and +// whether rep was the one registered. Two requests for the same unknown +// repository each build their own; the loser must be closed rather than left +// refreshing forever against nothing. +func (r *repositories) addIfAbsent(name string, rep appbase.Repository[[]byte]) (appbase.Repository[[]byte], bool) { + r.mu.Lock() + defer r.mu.Unlock() + if existing, ok := r.byName[name]; ok { + return existing, false + } + r.byName[name] = rep + return rep, true +} + +// snapshot copies the map so callers can range over it without holding the lock +// (and so /health reports one consistent view). +func (r *repositories) snapshot() map[string]appbase.Repository[[]byte] { + r.mu.RLock() + defer r.mu.RUnlock() + out := make(map[string]appbase.Repository[[]byte], len(r.byName)) + for name, rep := range r.byName { + out[name] = rep + } + return out } type RawRepositoryData struct { @@ -61,14 +113,12 @@ func (a *Context) InitContext(settings *appbase.AppSettings) error { refreshPeriodSec := a.config.RepositoryRefreshPeriodSec cacheDir := a.config.CacheDir - a.pScript = appbase.NewHTTPRepository[[]byte]("p.js", a.config.ScriptOrigin, "", appbase.HTTPTagETag, &RawRepositoryData{}, 5, 60, cacheDir) + a.pScript = appbase.NewHTTPRepository[[]byte]("p.js", a.config.ScriptOrigin, "", appbase.HTTPTagETag, &RawRepositoryData{}, 5, 60, cacheDir, appbase.WaitForData) reps := a.config.Repositories - a.repositories = map[string]appbase.Repository[[]byte]{ - "p.js": a.pScript, - } + a.repositories = newRepositories() + a.repositories.add("p.js", a.pScript) for _, rep := range strings.Split(reps, ",") { - a.repositories[rep] = appbase.NewHTTPRepository[[]byte](rep, baseUrl+"/"+rep, token, appbase.HTTPTagLastModified, &RawRepositoryData{validateJSON: true}, 2, refreshPeriodSec, cacheDir) - + a.repositories.add(rep, appbase.NewHTTPRepository[[]byte](rep, baseUrl+"/"+rep, token, appbase.HTTPTagLastModified, &RawRepositoryData{validateJSON: true}, 2, refreshPeriodSec, cacheDir, appbase.WaitForData)) } router := NewRouter(a) a.server = &http.Server{ @@ -82,7 +132,7 @@ func (a *Context) InitContext(settings *appbase.AppSettings) error { } func (a *Context) Cleanup() error { - for _, rep := range a.repositories { + for _, rep := range a.repositories.snapshot() { _ = rep.Close() } return nil diff --git a/bulker/config-keeper/repositories_test.go b/bulker/config-keeper/repositories_test.go new file mode 100644 index 000000000..1c772cc13 --- /dev/null +++ b/bulker/config-keeper/repositories_test.go @@ -0,0 +1,107 @@ +package main + +import ( + "fmt" + "sync" + "testing" + + "github.com/jitsucom/bulker/jitsubase/appbase" +) + +// stubRepository is enough to be stored in the map; the concurrency being tested +// is the map's, not the repository's. +type stubRepository struct { + appbase.Repository[[]byte] + closed bool +} + +func (s *stubRepository) Close() error { + s.closed = true + return nil +} + +// RepositoryHandler registers lazily discovered repositories from request +// goroutines while /health ranges over the map. On a plain map that combination +// aborts the process with "concurrent map read and map write". Run with -race. +func TestRepositoriesConcurrentAddAndSnapshot(t *testing.T) { + reps := newRepositories() + reps.add("preloaded", &stubRepository{}) + + const goroutines = 16 + var wg sync.WaitGroup + for i := 0; i < goroutines; i++ { + wg.Add(1) + go func(i int) { // writers, as the lazy-init path does + defer wg.Done() + reps.add(fmt.Sprintf("rep-%d", i), &stubRepository{}) + }(i) + wg.Add(1) + go func() { // readers, as /health does + defer wg.Done() + for name, rep := range reps.snapshot() { + if rep == nil { + t.Errorf("nil repository for %s", name) + } + } + }() + wg.Add(1) + go func(i int) { // point lookups, as RepositoryHandler does + defer wg.Done() + reps.get(fmt.Sprintf("rep-%d", i)) + }(i) + } + wg.Wait() + + if got := len(reps.snapshot()); got != goroutines+1 { + t.Errorf("got %d repositories, want %d", got, goroutines+1) + } +} + +// Two requests for the same unknown repository each build one. Exactly one may +// win, and the caller has to be told it lost so it can close the loser instead +// of leaving it refreshing forever. +func TestAddIfAbsentKeepsOneWinner(t *testing.T) { + reps := newRepositories() + + const goroutines = 8 + var wg sync.WaitGroup + winners := make([]bool, goroutines) + built := make([]*stubRepository, goroutines) + start := make(chan struct{}) + for i := 0; i < goroutines; i++ { + built[i] = &stubRepository{} + wg.Add(1) + go func(i int) { + defer wg.Done() + <-start + existing, stored := reps.addIfAbsent("contended", built[i]) + winners[i] = stored + if !stored { + _ = built[i].Close() + if existing == nil { + t.Error("addIfAbsent returned no repository for the loser") + } + } + }(i) + } + close(start) + wg.Wait() + + won := 0 + for i, w := range winners { + if w { + won++ + if built[i].closed { + t.Error("the winning repository was closed") + } + } else if !built[i].closed { + t.Error("a losing repository was left open") + } + } + if won != 1 { + t.Errorf("got %d winners, want exactly 1", won) + } + if got := len(reps.snapshot()); got != 1 { + t.Errorf("map holds %d entries, want 1", got) + } +} diff --git a/bulker/config-keeper/router.go b/bulker/config-keeper/router.go index e2c4ca65f..e5a7d5bc7 100644 --- a/bulker/config-keeper/router.go +++ b/bulker/config-keeper/router.go @@ -31,8 +31,10 @@ func NewRouter(appContext *Context) *Router { healthy := true repStatuses := map[string]any{} now := time.Now() + // one snapshot for both loops, so the two views cannot disagree + all := appContext.repositories.snapshot() for _, rep := range strings.Split(reps, ",") { - repository, ok := appContext.repositories[rep] + repository, ok := all[rep] if !ok { healthy = false repStatuses[rep] = map[string]any{"error": "not_found"} @@ -46,7 +48,7 @@ func NewRouter(appContext *Context) *Router { healthy = false } } - for name, repository := range appContext.repositories { + for name, repository := range all { lastSuccess := repository.LastSuccess() status := map[string]any{ "loaded": repository.Loaded(), @@ -74,32 +76,36 @@ func NewRouter(appContext *Context) *Router { } func (r *Router) RepositoryHandler(c *gin.Context) { repName := c.Param("repository") - repository, ok := r.appContext.repositories[repName] + repository, ok := r.appContext.repositories.get(repName) if !ok { r.Infof("Repository %s not found, initializing", repName) - repository = appbase.NewHTTPRepository[[]byte](repName, r.appContext.config.RepositoryBaseURL+"/"+repName, r.appContext.config.RepositoryAuthToken, appbase.HTTPTagLastModified, &RawRepositoryData{validateJSON: true}, 2, r.appContext.config.RepositoryRefreshPeriodSec, r.appContext.config.CacheDir) + repository = appbase.NewHTTPRepository[[]byte](repName, r.appContext.config.RepositoryBaseURL+"/"+repName, r.appContext.config.RepositoryAuthToken, appbase.HTTPTagLastModified, &RawRepositoryData{validateJSON: true}, 2, r.appContext.config.RepositoryRefreshPeriodSec, r.appContext.config.CacheDir, appbase.WaitForData) initTimeout := time.After(time.Second * 60) ticker := time.NewTicker(time.Second) defer ticker.Stop() - select { - case <-ticker.C: - if repository.Loaded() { - r.Infof("Repository %s initialized", repName) - r.appContext.repositories[repName] = repository - break - } - case <-initTimeout: - if !repository.Loaded() { - _ = repository.Close() - r.Errorf("Repository %s initialization timeout", repName) - _ = c.AbortWithError(http.StatusInternalServerError, fmt.Errorf("Repository %s initialization timeout", repName)) - return - } else { - r.Infof("Repository %s initialized", repName) - r.appContext.repositories[repName] = repository - break + // keep polling until it loads or the timeout fires: a single receive on + // the ticker only proves one second passed, not that the load finished + wait: + for !repository.Loaded() { + select { + case <-ticker.C: + case <-initTimeout: + break wait } } + if !repository.Loaded() { + _ = repository.Close() + r.Errorf("Repository %s initialization timeout", repName) + _ = c.AbortWithError(http.StatusInternalServerError, fmt.Errorf("repository %s initialization timeout", repName)) + return + } + r.Infof("Repository %s initialized", repName) + // a concurrent request for the same name may have finished first; keep + // whichever landed in the map and close the loser + if existing, stored := r.appContext.repositories.addIfAbsent(repName, repository); !stored { + _ = repository.Close() + repository = existing + } } var ifModifiedSince time.Time var err error @@ -110,6 +116,15 @@ func (r *Router) RepositoryHandler(c *gin.Context) { fmt.Println("Error parsing If-Modified-Since header:", err) } } + // Repositories wait for their datasource instead of killing the process, so + // one can be alive but never loaded - GetData is nil until the first load. + // Say so rather than dereference it, and let the consumer retry. + data := repository.GetData() + if data == nil { + r.Errorf("Repository %s is not loaded yet", repName) + c.AbortWithStatusJSON(http.StatusServiceUnavailable, gin.H{"error": fmt.Sprintf("repository %s is not loaded yet", repName)}) + return + } lastModified := repository.GetLastModified() if !ifModifiedSince.IsZero() && !lastModified.IsZero() && !lastModified.After(ifModifiedSince) { @@ -121,7 +136,7 @@ func (r *Router) RepositoryHandler(c *gin.Context) { c.Header("Last-Modified", lastModified.Format(http.TimeFormat)) } c.Writer.Header().Set("Content-Type", "application/json") - _, _ = c.Writer.Write(*repository.GetData()) + _, _ = c.Writer.Write(*data) } func (r *Router) ScriptHandler(c *gin.Context) { @@ -133,9 +148,15 @@ func (r *Router) ScriptHandler(c *gin.Context) { c.Status(http.StatusNotModified) return } + script := r.appContext.pScript.GetData() + if script == nil { + r.Errorf("p.js is not loaded yet") + c.AbortWithStatusJSON(http.StatusServiceUnavailable, gin.H{"error": "p.js is not loaded yet"}) + return + } if etag != "" { c.Header("ETag", etag) } c.Writer.Header().Set("Content-Type", "application/javascript") - _, _ = c.Writer.Write(*r.appContext.pScript.GetData()) + _, _ = c.Writer.Write(*script) } diff --git a/bulker/ingest/repository.go b/bulker/ingest/repository.go index 20f2bba86..002de0279 100644 --- a/bulker/ingest/repository.go +++ b/bulker/ingest/repository.go @@ -145,7 +145,7 @@ func (s *StreamsRepositoryData) Store(writer io.Writer) error { } func NewStreamsRepository(url, token string, refreshPeriodSec int, cacheDir string) appbase.Repository[Streams] { - return appbase.NewHTTPRepository[Streams]("streams-with-destinations", url, token, appbase.HTTPTagLastModified, &StreamsRepositoryData{}, 1, refreshPeriodSec, cacheDir) + return appbase.NewHTTPRepository[Streams]("streams-with-destinations", url, token, appbase.HTTPTagLastModified, &StreamsRepositoryData{}, 1, refreshPeriodSec, cacheDir, appbase.ExitOnNoData) } type DataLayout string diff --git a/bulker/ingest/script_repository.go b/bulker/ingest/script_repository.go index 66c8d796a..0efa686a4 100644 --- a/bulker/ingest/script_repository.go +++ b/bulker/ingest/script_repository.go @@ -97,5 +97,5 @@ func (s *ScriptRepositoryData) Store(writer io.Writer) error { } func NewScriptRepository(scriptOrigin, cacheDir string) *appbase.HTTPRepository[Script] { - return appbase.NewHTTPRepository[Script]("p.js", scriptOrigin, "", appbase.HTTPTagETag, &ScriptRepositoryData{}, 5, 120, cacheDir) + return appbase.NewHTTPRepository[Script]("p.js", scriptOrigin, "", appbase.HTTPTagETag, &ScriptRepositoryData{}, 5, 120, cacheDir, appbase.ExitOnNoData) } diff --git a/bulker/jitsubase/appbase/abstract_repository.go b/bulker/jitsubase/appbase/abstract_repository.go index 1caf28b97..494b932e9 100644 --- a/bulker/jitsubase/appbase/abstract_repository.go +++ b/bulker/jitsubase/appbase/abstract_repository.go @@ -25,6 +25,22 @@ type RepositoryData[D any] interface { Store(closer io.Writer) error } +// NoDataPolicy decides what a repository does when it has no data at all: the +// datasource is unreachable on a cold start and there is no usable cached copy. +type NoDataPolicy int + +const ( + // ExitOnNoData aborts the process. For services that would otherwise start + // serving traffic against empty configuration - an ingest that answers with + // an empty stream map rejects every event, which is worse than being down. + ExitOnNoData NoDataPolicy = iota + // WaitForData keeps retrying on the refresh ticker until the data appears. + // Loaded() stays false until then, so a consumer picking this MUST gate on + // it - both to keep itself out of the load balancer and to avoid reading + // data that is not there yet (GetData returns nil before the first load). + WaitForData +) + type AbstractRepository[T any] struct { Service changesChan chan bool @@ -37,13 +53,14 @@ type AbstractRepository[T any] struct { data RepositoryData[T] lastSuccess atomic.Pointer[time.Time] tag atomic.Pointer[any] + noDataPolicy NoDataPolicy closed chan struct{} } // RepositoryDataLoader loads data from external source. tag can be used for etag or last modified handling type RepositoryDataLoader func(tag any) (reader io.ReadCloser, newTag any, modified bool, err error) -func NewAbstractRepository[T any](id string, emptyData RepositoryData[T], source RepositoryDataLoader, attempts int, refreshPeriodSec int, cacheDir string) *AbstractRepository[T] { +func NewAbstractRepository[T any](id string, emptyData RepositoryData[T], source RepositoryDataLoader, attempts int, refreshPeriodSec int, cacheDir string, noDataPolicy NoDataPolicy) *AbstractRepository[T] { base := NewServiceBase(id) if attempts <= 0 { attempts = 1 @@ -56,31 +73,46 @@ func NewAbstractRepository[T any](id string, emptyData RepositoryData[T], source dataSource: source, attempts: attempts, data: emptyData, + noDataPolicy: noDataPolicy, closed: make(chan struct{}), } return r } +// noDataf reports that the repository has no usable data yet. Under ExitOnNoData +// it aborts the process; under WaitForData it logs and returns, leaving Loaded() +// false so the refresh ticker keeps trying until the datasource comes back. +// +// Only reached after a refresh has already failed, so every case it reports is +// genuinely abnormal - a missing cache file on its own is not an error. +func (r *AbstractRepository[T]) noDataf(format string, a ...any) { + if r.noDataPolicy == WaitForData { + r.Errorf(format+" Repository is not loaded, waiting for it to appear...", a...) + return + } + r.Fatalf(format+"\nCannot serve without repository. Exitting...", a...) +} + func (r *AbstractRepository[T]) loadCached() { file, err := os.Open(path.Join(r.cacheDir, r.ID)) if err != nil { - r.Fatalf("Error opening cached repository: %v\nCannot serve without repository. Exitting...", err) + r.noDataf("Error opening cached repository: %v.", err) return } defer file.Close() stat, err := file.Stat() if err != nil { - r.Fatalf("Error getting cached repository info: %v\nCannot serve without repository. Exitting...", err) + r.noDataf("Error getting cached repository info: %v.", err) return } fileSize := stat.Size() if fileSize == 0 { - r.Fatalf("Cached repository is empty\nCannot serve without repository. Exitting...") + r.noDataf("Cached repository is empty.") return } err = r.data.Init(file, nil) if err != nil { - r.Fatalf("Error init from cached repository: %v\nCannot serve without repository. Exitting...", err) + r.noDataf("Error init from cached repository: %v.", err) return } r.inited.Store(true) @@ -123,7 +155,7 @@ func (r *AbstractRepository[T]) refresh(notify bool) { if r.cacheDir != "" { r.loadCached() } else { - r.Fatalf("Cannot load cached repository. No CACHE_DIR is set. Cannot serve without repository. Exitting...") + r.noDataf("Cannot load cached repository: no CACHE_DIR is set.") } } } else { diff --git a/bulker/jitsubase/appbase/abstract_repository_test.go b/bulker/jitsubase/appbase/abstract_repository_test.go new file mode 100644 index 000000000..f7db671fa --- /dev/null +++ b/bulker/jitsubase/appbase/abstract_repository_test.go @@ -0,0 +1,146 @@ +package appbase + +import ( + "fmt" + "io" + "strings" + "sync/atomic" + "testing" + + log "github.com/sirupsen/logrus" +) + +type testData struct { + data atomic.Pointer[string] +} + +func (t *testData) Init(reader io.Reader, tag any) error { + b, err := io.ReadAll(reader) + if err != nil { + return err + } + s := string(b) + t.data.Store(&s) + return nil +} + +func (t *testData) GetData() *string { return t.data.Load() } + +func (t *testData) Store(w io.Writer) error { + d := t.data.Load() + if d == nil { + return fmt.Errorf("no data") + } + _, err := w.Write([]byte(*d)) + return err +} + +// captureExit swallows the process exit that logging.Fatalf performs and reports +// whether it was reached. +func captureExit(t *testing.T) *atomic.Bool { + t.Helper() + var exited atomic.Bool + logger := log.StandardLogger() + prevExit, prevOut := logger.ExitFunc, logger.Out + logger.ExitFunc = func(int) { exited.Store(true) } + logger.SetOutput(io.Discard) + t.Cleanup(func() { + logger.ExitFunc = prevExit + logger.SetOutput(prevOut) + }) + return &exited +} + +// failingSource fails until it is switched on, then serves payload. +type failingSource struct { + ok atomic.Bool + payload string +} + +func (f *failingSource) load(tag any) (io.ReadCloser, any, bool, error) { + if !f.ok.Load() { + return nil, nil, false, fmt.Errorf("datasource is down") + } + return io.NopCloser(strings.NewReader(f.payload)), "tag", true, nil +} + +// A repository that has never loaded must keep waiting rather than kill the +// process - config-keeper serves several repositories from one process, so an +// exit over one unreachable datasource takes down the ones that are healthy. +func TestWaitForDataDoesNotExit(t *testing.T) { + exited := captureExit(t) + src := &failingSource{payload: "hello"} + // cacheDir "" is the harshest case: no datasource and nothing cached + r := NewAbstractRepository[string]("test-wait", &testData{}, src.load, 1, 1, "", WaitForData) + + r.refresh(false) + + if exited.Load() { + t.Fatal("WaitForData repository exited the process on a failed initial load") + } + if r.Loaded() { + t.Error("Loaded() must stay false while there is no data") + } + if r.GetData() != nil { + t.Error("GetData() must be nil before the first successful load") + } + + // ...and it must pick the data up once the datasource comes back + src.ok.Store(true) + r.refresh(false) + + if !r.Loaded() { + t.Fatal("repository did not load after the datasource recovered") + } + if got := r.GetData(); got == nil || *got != "hello" { + t.Errorf("got %v, want \"hello\"", got) + } + if exited.Load() { + t.Error("process exited during recovery") + } +} + +// The default stays fail-fast: a service that would otherwise serve traffic +// against empty configuration should not come up at all. +func TestExitOnNoDataStillExits(t *testing.T) { + exited := captureExit(t) + src := &failingSource{payload: "hello"} + r := NewAbstractRepository[string]("test-exit", &testData{}, src.load, 1, 1, "", ExitOnNoData) + + r.refresh(false) + + if !exited.Load() { + t.Error("ExitOnNoData repository did not exit on a failed initial load") + } +} + +// Once loaded, a failing refresh must never reach the no-data path under either +// policy: the previous good data keeps being served. +func TestLoadedRepositoryKeepsServingAfterRefreshFailure(t *testing.T) { + for _, policy := range []struct { + name string + value NoDataPolicy + }{{"WaitForData", WaitForData}, {"ExitOnNoData", ExitOnNoData}} { + t.Run(policy.name, func(t *testing.T) { + exited := captureExit(t) + src := &failingSource{payload: "hello"} + src.ok.Store(true) + r := NewAbstractRepository[string]("test-"+policy.name, &testData{}, src.load, 1, 1, "", policy.value) + + r.refresh(false) + if !r.Loaded() { + t.Fatal("setup: repository did not load") + } + + src.ok.Store(false) + r.refresh(false) + + if exited.Load() { + t.Error("a refresh failure after a successful load must not exit") + } + if got := r.GetData(); got == nil || *got != "hello" { + t.Errorf("previous data was lost: got %v", got) + } + }) + } +} diff --git a/bulker/jitsubase/appbase/http_repository.go b/bulker/jitsubase/appbase/http_repository.go index 26d06c387..f0e4e183f 100644 --- a/bulker/jitsubase/appbase/http_repository.go +++ b/bulker/jitsubase/appbase/http_repository.go @@ -22,8 +22,8 @@ type HTTPRepository[T any] struct { tagHeader CacheTagHeader } -func NewHTTPRepository[T any](id, url, token string, tagHeader CacheTagHeader, emptyData RepositoryData[T], attempts int, refreshPeriodSec int, cacheDir string) *HTTPRepository[T] { - a := NewAbstractRepository[T](id, emptyData, nil, attempts, refreshPeriodSec, cacheDir) +func NewHTTPRepository[T any](id, url, token string, tagHeader CacheTagHeader, emptyData RepositoryData[T], attempts int, refreshPeriodSec int, cacheDir string, noDataPolicy NoDataPolicy) *HTTPRepository[T] { + a := NewAbstractRepository[T](id, emptyData, nil, attempts, refreshPeriodSec, cacheDir, noDataPolicy) r := &HTTPRepository[T]{ AbstractRepository: a, url: url, diff --git a/bulker/operator/repository.go b/bulker/operator/repository.go index 492af56b8..20c7a8f63 100644 --- a/bulker/operator/repository.go +++ b/bulker/operator/repository.go @@ -221,17 +221,17 @@ func (w *WorkspacesRepositoryData) Store(writer io.Writer) error { // Repository factory functions func NewConnectionsRepository(baseURL, token string, refreshPeriodSec int, cacheDir string) appbase.Repository[ConnectionsData] { url := fmt.Sprintf("%s/rotor-connections", baseURL) - return appbase.NewHTTPRepository[ConnectionsData]("rotor-connections", url, token, appbase.HTTPTagLastModified, &ConnectionsRepositoryData{}, 1, refreshPeriodSec, cacheDir) + return appbase.NewHTTPRepository[ConnectionsData]("rotor-connections", url, token, appbase.HTTPTagLastModified, &ConnectionsRepositoryData{}, 1, refreshPeriodSec, cacheDir, appbase.WaitForData) } func NewFunctionsRepository(baseURL, token string, refreshPeriodSec int, cacheDir string) appbase.Repository[FunctionsData] { url := fmt.Sprintf("%s/functions", baseURL) - return appbase.NewHTTPRepository[FunctionsData]("functions", url, token, appbase.HTTPTagLastModified, &FunctionsRepositoryData{}, 1, refreshPeriodSec, cacheDir) + return appbase.NewHTTPRepository[FunctionsData]("functions", url, token, appbase.HTTPTagLastModified, &FunctionsRepositoryData{}, 1, refreshPeriodSec, cacheDir, appbase.WaitForData) } func NewWorkspacesRepository(baseURL, token string, refreshPeriodSec int, cacheDir string) appbase.Repository[WorkspacesData] { url := fmt.Sprintf("%s/workspaces-with-profiles", baseURL) - return appbase.NewHTTPRepository[WorkspacesData]("workspaces-with-profiles", url, token, appbase.HTTPTagLastModified, &WorkspacesRepositoryData{}, 1, refreshPeriodSec, cacheDir) + return appbase.NewHTTPRepository[WorkspacesData]("workspaces-with-profiles", url, token, appbase.HTTPTagLastModified, &WorkspacesRepositoryData{}, 1, refreshPeriodSec, cacheDir, appbase.WaitForData) } // Helper functions for aggregating workspace data diff --git a/bulker/reprocessing-worker/repository.go b/bulker/reprocessing-worker/repository.go index 67cbe4281..399f1dfba 100644 --- a/bulker/reprocessing-worker/repository.go +++ b/bulker/reprocessing-worker/repository.go @@ -84,7 +84,7 @@ func (s *StreamsRepositoryData) Store(writer io.Writer) error { } func NewStreamsRepository(url, token string, refreshPeriodSec int, cacheDir string) appbase.Repository[Streams] { - return appbase.NewHTTPRepository[Streams]("streams-with-destinations", url, token, appbase.HTTPTagLastModified, &StreamsRepositoryData{}, 1, refreshPeriodSec, cacheDir) + return appbase.NewHTTPRepository[Streams]("streams-with-destinations", url, token, appbase.HTTPTagLastModified, &StreamsRepositoryData{}, 1, refreshPeriodSec, cacheDir, appbase.ExitOnNoData) } type StreamConfig struct { diff --git a/bulker/sync-controller/repository.go b/bulker/sync-controller/repository.go index 9c9a5b7a8..4a4ba63bf 100644 --- a/bulker/sync-controller/repository.go +++ b/bulker/sync-controller/repository.go @@ -110,7 +110,7 @@ func (s *SyncsRepositoryData) Store(writer io.Writer) error { // NewSyncsRepository wires the syncs export polling repository. func NewSyncsRepository(baseURL, token string, refreshPeriodSec int, cacheDir string) appbase.Repository[SyncsData] { url := fmt.Sprintf("%s/syncs", baseURL) - return appbase.NewHTTPRepository[SyncsData]("syncs", url, token, appbase.HTTPTagLastModified, &SyncsRepositoryData{}, 1, refreshPeriodSec, cacheDir) + return appbase.NewHTTPRepository[SyncsData]("syncs", url, token, appbase.HTTPTagLastModified, &SyncsRepositoryData{}, 1, refreshPeriodSec, cacheDir, appbase.WaitForData) } // WaitForSyncEntry blocks until the repository contains a SyncEntry for syncID