Count the take-ins and every other background work in the serving group, so serve's stop waits for them too (issue 1945)

inBackground ran the take-ins of done keys (now in #463's slots), the start's
look, the given values' replacement and the announced reads beside serve without
serve waiting for them: a take-in could outlive its controller as the loops did.
It now takes the context and is counted in its serving group as well, and a
serve that is stopping starts none.
This commit is contained in:
2026-10-25 16:36:49 +01:00
parent c7e31640cf
commit cc5c9ae888
6 changed files with 22 additions and 18 deletions
+3 -3
View File
@@ -307,7 +307,7 @@ func watchDoneKeys(ctx context.Context, open *stores) {
return
}
if err := builderkeys.Watch(ctx, keys.JS, func(module string) {
inBackground(func() { takeDoneKey(ctx, open, module) })
inBackground(ctx, func() { takeDoneKey(ctx, open, module) })
}); err != nil {
closeKeys()
fmt.Printf("done keys: the builder's keys are not watched: %v\n", err)
@@ -318,7 +318,7 @@ func watchDoneKeys(ctx context.Context, open *stores) {
closeKeys()
})
// And every module once now, after the watch is open so no write falls between the two.
inBackground(func() {
inBackground(ctx, func() {
doneKeysAtStart(ctx, catalogedModules(open), func(module string) { takeDoneKeyAtStart(ctx, open, module) })
})
}
@@ -389,7 +389,7 @@ func takeDoneKeyAtStart(ctx context.Context, open *stores, module string) {
_, again := doneKeysLookAgain.LoadAndDelete(module)
if err != nil || again {
// A failure, a wait or a wake meanwhile is followed by takeDoneKey, which says it and looks again.
inBackground(func() { takeDoneKey(ctx, open, module) })
inBackground(ctx, func() { takeDoneKey(ctx, open, module) })
}
}
+1 -1
View File
@@ -152,7 +152,7 @@ func watchHavesForJudgement(ctx context.Context, open *stores) {
if err != nil || isNode {
continue
}
inBackground(func() {
inBackground(ctx, func() {
if err := rejudgeOn(ctx, open, []string{node}, module); err != nil {
fmt.Printf("compositions: %s on %s is not judged again: %v\n", module, node, err)
}
+1 -1
View File
@@ -195,7 +195,7 @@ func watchHavesForGiven(ctx context.Context, open *stores) {
if at.IsZero() {
continue
}
inBackground(func() { replaceGiven(ctx, open, node, map[string]time.Time{module: at}) })
inBackground(ctx, func() { replaceGiven(ctx, open, node, map[string]time.Time{module: at}) })
}
}
})
+11 -8
View File
@@ -200,13 +200,13 @@ func newModuleObservations(now time.Time) []conditions.Observation {
// watchNewModules wakes the take-in of new modules for each module the builder or the indexer writes, until ctx ends, and
// looks at every module once now. A bucket not there yet is said: the look at start and a first merge's word still work.
func watchNewModules(ctx context.Context, open *stores) {
inBackground(func() {
inBackground(ctx, func() {
takeNewModules(ctx, open)
})
// And, beside it, the start's pass of release 6 and the take-in of each registered module's new done key (novox/hq
// ADR 0539 clauses 5, 7, 10 and 13; composed_start.go, composed_keys.go): the plans' buckets deleted, every node
// composed, the replaced controller composed, and from now on each new fingerprint registered, composed and granted.
inBackground(func() { composeAtStart(ctx, open) })
inBackground(ctx, func() { composeAtStart(ctx, open) })
watchDoneKeys(ctx, open)
watchHavesForJudgement(ctx, open)
// And the keep set keeps the builds every want names (ADR 0539 clause 1), with no plan left to name them.
@@ -232,7 +232,7 @@ func watchNewModules(ctx context.Context, open *stores) {
// wakeNewModule takes module in when the catalogue does not hold it: never on the watch's own goroutine, which must keep
// draining.
func wakeNewModule(ctx context.Context, open *stores, module string) {
inBackground(func() {
inBackground(ctx, func() {
entries, err := open.inventory.Catalogued(ctx)
if err != nil {
return
@@ -250,14 +250,17 @@ func wakeNewModule(ctx context.Context, open *stores, module string) {
// ends only once the work its mesh started has.
var background sync.WaitGroup
// inBackground runs work beside the caller, counted in background. A variable so a test of what a walk does around a
// take-in may run it in the caller's turn and need not wait for it.
var inBackground = func(work func()) {
// inBackground runs work beside the caller, counted in background and, under a serving controller, in its serving group,
// so serve's stop waits for it and a serve that is stopping starts none (serving_group.go). A variable so a test of what a
// walk does around a take-in may run it in the caller's turn and need not wait for it.
var inBackground = func(ctx context.Context, work func()) {
background.Add(1)
go func() {
if !goServing(ctx, func() {
defer background.Done()
work()
}()
}) {
background.Done()
}
}
// newModulesOf are the modules the indexer holds in a repository: what a first merge's word wakes.
+5 -4
View File
@@ -29,17 +29,17 @@ func withServingGroup(ctx context.Context) (context.Context, *servingGroup) {
// goServing runs work beside the caller, counted in the group ctx carries, so serve's stop waits for it. Under a group
// whose stop began it runs nothing; under a context with no group (a part run alone, by a test or a verb) it is a plain
// goroutine, as before.
func goServing(ctx context.Context, work func()) {
// goroutine, as before. It says whether it started work.
func goServing(ctx context.Context, work func()) bool {
g, _ := ctx.Value(servingGroupKey{}).(*servingGroup)
if g == nil {
go work()
return
return true
}
g.mu.Lock()
if g.stopped {
g.mu.Unlock()
return
return false
}
g.wg.Add(1)
g.mu.Unlock()
@@ -47,6 +47,7 @@ func goServing(ctx context.Context, work func()) {
defer g.wg.Done()
work()
}()
return true
}
// stop starts nothing more and waits for what was started; the caller ends their context first. A second stop waits
+1 -1
View File
@@ -67,7 +67,7 @@ func keepAnnouncements(ctx context.Context, js *broker.JetStream, open *stores)
return
}
if changed {
inBackground(func() { composeAnnouncedReads(context.WithoutCancel(ctx), open) })
inBackground(ctx, func() { composeAnnouncedReads(context.WithoutCancel(ctx), open) })
}
})
if err != nil {