From cc5c9ae888929dfd29a4b9d9d006bc5d5eac1df8 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 16:36:49 +0100 Subject: [PATCH] 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. --- cmd/mesh-controller/composed_keys.go | 6 +++--- cmd/mesh-controller/composed_rejudge.go | 2 +- cmd/mesh-controller/given.go | 2 +- cmd/mesh-controller/new_modules.go | 19 +++++++++++-------- cmd/mesh-controller/serving_group.go | 9 +++++---- cmd/mesh-controller/states_announced.go | 2 +- 6 files changed, 22 insertions(+), 18 deletions(-) diff --git a/cmd/mesh-controller/composed_keys.go b/cmd/mesh-controller/composed_keys.go index c2d31118..6ae5eea9 100644 --- a/cmd/mesh-controller/composed_keys.go +++ b/cmd/mesh-controller/composed_keys.go @@ -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) }) } } diff --git a/cmd/mesh-controller/composed_rejudge.go b/cmd/mesh-controller/composed_rejudge.go index 09997ffa..73215784 100644 --- a/cmd/mesh-controller/composed_rejudge.go +++ b/cmd/mesh-controller/composed_rejudge.go @@ -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) } diff --git a/cmd/mesh-controller/given.go b/cmd/mesh-controller/given.go index 83dfa50a..1d905b01 100644 --- a/cmd/mesh-controller/given.go +++ b/cmd/mesh-controller/given.go @@ -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}) }) } } }) diff --git a/cmd/mesh-controller/new_modules.go b/cmd/mesh-controller/new_modules.go index e52b695a..098d9404 100644 --- a/cmd/mesh-controller/new_modules.go +++ b/cmd/mesh-controller/new_modules.go @@ -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. diff --git a/cmd/mesh-controller/serving_group.go b/cmd/mesh-controller/serving_group.go index cf054fa6..f810be0c 100644 --- a/cmd/mesh-controller/serving_group.go +++ b/cmd/mesh-controller/serving_group.go @@ -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 diff --git a/cmd/mesh-controller/states_announced.go b/cmd/mesh-controller/states_announced.go index f68cdb0e..dbd17204 100644 --- a/cmd/mesh-controller/states_announced.go +++ b/cmd/mesh-controller/states_announced.go @@ -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 {