fix(cfgkpr): wait for a repository to appear instead of exiting

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}
 }
 

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

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)
+	}
+}

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)
 }

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

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)
 }