diff --git a/cmd/mesh-controller/first_builds.go b/cmd/mesh-controller/first_builds.go index c6f6322c..b6c06bfc 100644 --- a/cmd/mesh-controller/first_builds.go +++ b/cmd/mesh-controller/first_builds.go @@ -94,7 +94,7 @@ func askFirstBuilds(ctx context.Context, open *stores, m link.SourceMoved, now t fmt.Printf(" the indexer's modules of %s cannot be read now (%v): they are registered when the builder's next "+ "write of one wakes the take-in\n", repository, err) case len(modules) > 0: - go takeNewModules(context.WithoutCancel(ctx), open, modules...) + inBackground(func() { takeNewModules(context.WithoutCancel(ctx), open, modules...) }) } return nil } diff --git a/cmd/mesh-controller/mesh_for_test.go b/cmd/mesh-controller/mesh_for_test.go index 43f73c75..4cfe3ff7 100644 --- a/cmd/mesh-controller/mesh_for_test.go +++ b/cmd/mesh-controller/mesh_for_test.go @@ -56,6 +56,8 @@ func aMeshOnTheStore(t *testing.T) *stores { t.Fatal(err) } t.Cleanup(open.Close) + // The work this mesh started in the background ends before its stores close (cleanups run last first). + t.Cleanup(background.Wait) // A condition store of its own, held in memory (novox/hq to-be 45 ยง2): status leads with what is // open, and a mesh with no bus would otherwise read as one whose conditions cannot be read. withConditionsInMemory(t) diff --git a/cmd/mesh-controller/new_modules.go b/cmd/mesh-controller/new_modules.go index 58d88cd3..aa870fe7 100644 --- a/cmd/mesh-controller/new_modules.go +++ b/cmd/mesh-controller/new_modules.go @@ -200,10 +200,10 @@ 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) { - go func() { + inBackground(func() { answerHeldMerges(ctx, open.inventory) takeNewModules(ctx, open) - }() + }) keys, closeKeys, err := openBuilderKeys() if err != nil { fmt.Printf("new modules: the builder's keys are not watched: %v\n", err) @@ -223,7 +223,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) { - go func() { + inBackground(func() { entries, err := open.inventory.Catalogued(ctx) if err != nil { return @@ -236,6 +236,19 @@ func wakeNewModule(ctx context.Context, open *stores, module string) { } } takeNewModules(ctx, open, module) + }) +} + +// background is the work the take-ins of keys and of announcements start beside the event that woke them, so a test +// ends only once the work its mesh started has. +var background sync.WaitGroup + +// inBackground runs work beside the caller, counted in background. +func inBackground(work func()) { + background.Add(1) + go func() { + defer background.Done() + work() }() } diff --git a/cmd/mesh-controller/plan_keys.go b/cmd/mesh-controller/plan_keys.go index f3fe8d35..c3677e6e 100644 --- a/cmd/mesh-controller/plan_keys.go +++ b/cmd/mesh-controller/plan_keys.go @@ -73,7 +73,7 @@ var keyTakes = struct { // the buckets and the records; a done key not recorded yet is taken in off the plans' hold, and its end asks the look. func settleFromKeys(ctx context.Context, open *stores, keys planKeyReads, p *inventory.Plan, tier []string) (bool, error) { return settleKeys(ctx, open.inventory, keys, p, tier, func(module string, trunk sdkdelivery.Module, id string) { - go takeKeyOffTheHold(context.WithoutCancel(ctx), open, module, trunk, id) + inBackground(func() { takeKeyOffTheHold(context.WithoutCancel(ctx), open, module, trunk, id) }) }) } diff --git a/cmd/mesh-controller/states_announced.go b/cmd/mesh-controller/states_announced.go index 8e4012a1..f4f05611 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 { - go composeAnnouncedReads(context.WithoutCancel(ctx), open) + inBackground(func() { composeAnnouncedReads(context.WithoutCancel(ctx), open) }) } }) if err != nil {