fix: prevent race by forcing nobori to check ready

Author: madonuko

gura.go

@@ -23,8 +23,8 @@ func main() {
 	}
 
 	db.SetupDB()
+	nobori.StartFetchLoop()
 	go kudari.FetchLoop()
-	go nobori.FetchLoop()
 
 	router := api.SetupRouter(router_opts)
 	var listen_address = ":8080"

nobori/github.go

@@ -168,9 +168,9 @@ func ghFillTokQl() *GHToken {
 			})
 		var q struct {
 			RateLimit struct {
-				Remaining int16  `graphql:"remaining"`
-				ResetAt   string `graphql:"resetAt"`
-			} `graphql:"rateLimit"`
+				Remaining int16
+				ResetAt   string
+			}
 		}
 		err := qlcli.Query(context.Background(), &q, nil)
 		if err != nil {
@@ -190,14 +190,22 @@ func ghFillTokQl() *GHToken {
 // Swimming
 
 // the shark shall initialise the funny
-func GhSwim() {
+func GhSwim() chan struct{} {
 	go ghSwimRt()
 	go ghSwimQl()
+	ready := make(chan struct{}, 1)
+	go func() {
+		for len(ghTokRt) == 0 || len(ghTokQl) == 0 {
+			time.Sleep(20 * time.Millisecond)
+		}
+		ready <- struct{}{}
+	}()
+	return ready
 }
 
 // Process GitHub REST streams using available tokens
 func ghSwimRt() {
-	for token := ghFillTokRt(); token.noMoreFish(); token.thanksForAllTheFish(&ghRtTokIdx, &ghTokRt) {
+	for token := ghFillTokRt(); ; token.thanksForAllTheFish(&ghRtTokIdx, &ghTokRt) {
 		token.waitForFish()
 		for !token.noMoreFish() {
 			stream := <-ghRtPool
@@ -211,7 +219,7 @@ func ghSwimRt() {
 
 // Process GitHub GraphQL streams using available tokens
 func ghSwimQl() {
-	for token := ghFillTokQl(); token.noMoreFish(); token.thanksForAllTheFish(&ghQlTokIdx, &ghTokQl) {
+	for token := ghFillTokQl(); ; token.thanksForAllTheFish(&ghQlTokIdx, &ghTokQl) {
 		token.waitForFish()
 		qlcli := token.qlcli()
 		for {

nobori/nobori.go

@@ -23,6 +23,11 @@ var l = util.SetupLog("nobori")
 // Central channel for scheduling upstream stream fetches
 var queue chan db.Stream = make(chan db.Stream)
 
+// Upstream handlers
+// Each handler function should return a channel.
+// Sending a message to the channel indicates the swimmer is ready.
+var swimmers = []func() chan struct{}{GhSwim}
+
 // Calculate the timeout duration
 //
 // A power-law function is used to calculate the timeout duration.
@@ -49,14 +54,24 @@ func schedule(stream db.Stream) {
 	queue <- stream
 }
 
-// Main loop for upstream metadata fetching
+// Start the fetch loop for upstream metadata fetching
 //
-// Streams are processed from the queue as their scheduled time arrives.
-func FetchLoop() {
-	go GhSwim()
+// Return when nobori is ready.
+func StartFetchLoop() {
+	ready_chs := util.SliceMap(swimmers, func(swimmer func() chan struct{}) chan struct{} { return swimmer() })
 	var strms []db.Stream
 	r := db.DB.Find(&strms)
 	util.Yeet(l, "can't find streams", r.Error)
+
+	util.SliceEach(ready_chs, func(ch chan struct{}) { <-ch })
+
+	go fetchLoop(strms)
+}
+
+// Main loop for upstream metadata fetching
+//
+// Streams are processed from the queue as their scheduled time arrives.
+func fetchLoop(strms []db.Stream) {
 	for _, stream := range strms {
 		go schedule(stream)
 	}

nobori/strm.go

@@ -21,6 +21,7 @@ var newStrmHdlrs = []func(db *gorm.DB, p *db.Pkg, url string, urls []string) boo
 		if !strings.HasPrefix(url, "github.com/") {
 			return false
 		}
+		l.Debug("strm hdl github", zap.String("p", p.ID.String()))
 		strm := GhStrmHdlr(*p, url)
 		strm.Mirrors = strings.Join(urls, ",")
 		dbx.Save(strm)
@@ -83,6 +84,7 @@ recv:
 }
 
 func handleNewStrm(dbx *gorm.DB, p *db.Pkg, urls []string) error {
+	l.Debug("new strm", zap.String("ID", p.ID.String()))
 	for _, hdlr := range newStrmHdlrs {
 		for _, url := range urls {
 			if hdlr(dbx, p, url, urls) {

util/sugar.go

@@ -10,3 +10,9 @@ func SliceMap[T, U any](slice []T, fn func(T) U) []U {
 	}
 	return result
 }
+
+func SliceEach[T any](slice []T, fn func(T)) {
+	for _, v := range slice {
+		fn(v)
+	}
+}