diff --git a/cmd/mesh-controller/asker_wire.go b/cmd/mesh-controller/asker_wire.go index 59c75efa..7f756226 100644 --- a/cmd/mesh-controller/asker_wire.go +++ b/cmd/mesh-controller/asker_wire.go @@ -12,6 +12,7 @@ import ( "sort" "strings" "sync" + "sync/atomic" "time" "github.com/nats-io/nats.go" @@ -27,7 +28,7 @@ import ( ) // askerFrom is the serving controller's asker; nil in any other process. -var askerFrom *asker +var askerFrom atomic.Pointer[asker] // askWithin is how long a verb a warrant chose is given to answer. const askWithin = time.Minute @@ -573,8 +574,8 @@ func startAsking(ctx context.Context, open *stores, server *link.Server, conn *n fmt.Printf("the operator's answers cannot be heard, so nothing is asked: %v\n", err) return } - askerFrom = a - go a.keep(ctx) + askerFrom.Store(a) + goServing(ctx, func() { a.keep(ctx) }) } // servingAsker is the serving controller's asker, wired to the mesh: what it reads, publishes, performs and diff --git a/cmd/mesh-controller/chain.go b/cmd/mesh-controller/chain.go index 5c58ff7a..582fcff1 100644 --- a/cmd/mesh-controller/chain.go +++ b/cmd/mesh-controller/chain.go @@ -49,7 +49,7 @@ func attachTheChain(ctx context.Context, inv *inventory.Inventory, server *link. // keys laid out before skips them, so this takes seconds; one that fails is tried again beside the serving, loudly. if err := inv.LayBuildsOut(ctx, func(format string, args ...any) { fmt.Printf("chain: "+format+"\n", args...) }); err != nil { fmt.Printf("chain: the builds are not laid out under their modules yet, and none is read until they are: %v\n", err) - go layBuildsOutSoon(ctx, inv) + goServing(ctx, func() { layBuildsOutSoon(ctx, inv) }) } // The requests the move of delivery 1 found never handed over are not asked again: the controller asks no build // (novox/hq ADR 0501; issue 1851, A10), and the builder builds every fingerprint the indexer writes. diff --git a/cmd/mesh-controller/composed_keys.go b/cmd/mesh-controller/composed_keys.go index 8b8dd3ff..36a4a2cb 100644 --- a/cmd/mesh-controller/composed_keys.go +++ b/cmd/mesh-controller/composed_keys.go @@ -194,6 +194,9 @@ func followDoneKey(ctx context.Context, open *stores, module string) { } continue } + if ctx.Err() != nil { + return // stopped: a take-in cut by the stop is no fault of the build, and the next controller takes it in + } if keeper := conditionsFrom; keeper != nil && !failed { o.Summary += err.Error() if _, oerr := keeper.Observe(ctx, o); oerr == nil { @@ -308,10 +311,10 @@ func watchDoneKeys(ctx context.Context, open *stores) { fmt.Printf("done keys: the builder's keys are not watched: %v\n", err) return } - go func() { + goServing(ctx, func() { <-ctx.Done() closeKeys() - }() + }) // And every module once now, after the watch is open so no write falls between the two. inBackground(func() { doneKeysAtStart(ctx, catalogedModules(open), func(module string) { takeDoneKeyAtStart(ctx, open, module) }) diff --git a/cmd/mesh-controller/composed_rejudge.go b/cmd/mesh-controller/composed_rejudge.go index 0393123c..09997ffa 100644 --- a/cmd/mesh-controller/composed_rejudge.go +++ b/cmd/mesh-controller/composed_rejudge.go @@ -134,7 +134,7 @@ func watchHavesForJudgement(ctx context.Context, open *stores) { fmt.Printf("compositions: the haves are not watched: %v\n", err) return } - go func() { + goServing(ctx, func() { defer bus.Close() defer w.Stop() for { @@ -165,5 +165,5 @@ func watchHavesForJudgement(ctx context.Context, open *stores) { }) } } - }() + }) } diff --git a/cmd/mesh-controller/composed_served_integration_test.go b/cmd/mesh-controller/composed_served_integration_test.go index 56a5abe4..dc53e041 100644 --- a/cmd/mesh-controller/composed_served_integration_test.go +++ b/cmd/mesh-controller/composed_served_integration_test.go @@ -608,8 +608,11 @@ func keepWhatServeSets(t *testing.T) { t.Setenv(servedVar, "") unheld, status, given, check, follow, keeper, hand := logUnheldChanges, statusFrom, givenEvents, checkEvents, link.Calls.Follow, conditionsFrom, handActConn + doctor, asker := doctorFrom, askerFrom.Load() t.Cleanup(func() { logUnheldChanges, statusFrom, givenEvents, checkEvents, link.Calls.Follow, conditionsFrom, handActConn = unheld, status, given, check, follow, keeper, hand + doctorFrom = doctor + askerFrom.Store(asker) }) } diff --git a/cmd/mesh-controller/conditions.go b/cmd/mesh-controller/conditions.go index 8c64b3a9..4f70a21c 100644 --- a/cmd/mesh-controller/conditions.go +++ b/cmd/mesh-controller/conditions.go @@ -49,7 +49,7 @@ func keeperOn(ctx context.Context, conn *nats.Conn, looks conditions.LookStore) Say: func(format string, args ...any) { fmt.Fprintf(os.Stderr, format+"\n", args...) }, // What status leads with changed: composed again soon (a nudge outside the serving controller // does nothing). - Changed: func() { statusFrom.nudge(); askerFrom.nudge() }, + Changed: func() { statusFrom.nudge(); askerFrom.Load().nudge() }, // Written under the lease, carrying its epoch (novox/hq to-be 45 §6). Epoch: func() (uint64, error) { return theLease.epoch(context.WithoutCancel(ctx)) }}), nil } diff --git a/cmd/mesh-controller/given.go b/cmd/mesh-controller/given.go index 5f240b57..83dfa50a 100644 --- a/cmd/mesh-controller/given.go +++ b/cmd/mesh-controller/given.go @@ -169,7 +169,7 @@ func watchHavesForGiven(ctx context.Context, open *stores) { log.Printf("given values: the haves are not watched: %v", err) return } - go func() { + goServing(ctx, func() { defer bus.Close() defer w.Stop() for { @@ -198,5 +198,5 @@ func watchHavesForGiven(ctx context.Context, open *stores) { inBackground(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 0273ff3d..e52b695a 100644 --- a/cmd/mesh-controller/new_modules.go +++ b/cmd/mesh-controller/new_modules.go @@ -223,10 +223,10 @@ func watchNewModules(ctx context.Context, open *stores) { fmt.Printf("new modules: the builder's keys are not watched: %v\n", err) return } - go func() { + goServing(ctx, func() { <-ctx.Done() closeKeys() - }() + }) } // wakeNewModule takes module in when the catalogue does not hold it: never on the watch's own goroutine, which must keep diff --git a/cmd/mesh-controller/plain_words.go b/cmd/mesh-controller/plain_words.go index b6b73bdb..2b1fb1b0 100644 --- a/cmd/mesh-controller/plain_words.go +++ b/cmd/mesh-controller/plain_words.go @@ -59,6 +59,17 @@ func dataWords(o conditions.Observation) (what, module string) { } var plainWordings = map[string]func(conditions.Observation) words{ + // A module's new build whose compositions or grant the controller could not write (composed_keys.go). + kindNotComposed: worded(func(o conditions.Observation) words { + m := "a module" + if p := idPart(o, 0); o.Scope == conditions.ScopeBuild && p != "" { + m = p // the module: raised in the build's scope, its id the module's name + } + return words{Headline: "A new build of " + m + " cannot be delivered", + Explanation: fmt.Sprintf("A new build of %s was made, and the mesh could not prepare it for the machines, so "+ + "none can be given it yet. It is tried again every half minute; the details say what fails.", m), + Resolved: "the new build of " + m + " can be delivered"} + }), // Machines and their node-engine. "silent": worded(func(o conditions.Observation) words { m := machineOr(o, "a machine") diff --git a/cmd/mesh-controller/push.go b/cmd/mesh-controller/push.go index 7c63572d..7d18c351 100644 --- a/cmd/mesh-controller/push.go +++ b/cmd/mesh-controller/push.go @@ -125,7 +125,15 @@ func serve(ctx context.Context) (err error) { defer theLease.release() ctx, stopActing := context.WithCancel(ctx) defer stopActing() - go func() { + // Every goroutine started from here is counted, and ended and waited for before serve returns (serving_group.go): + // on the way out by error here, and at the end before anything else is undone. + ctx, running := withServingGroup(ctx) + stopRunning := func() { + stopActing() + running.stop() + } + defer stopRunning() + goServing(ctx, func() { select { case <-lost: stopActing() @@ -135,7 +143,7 @@ func serve(ctx context.Context) (err error) { // asks the controller after this one rather than have a command started and killed with this // process (novox/hq issue 289). handingOver.Store(true) - }() + }) defer func() { select { case <-lost: @@ -162,6 +170,16 @@ func serve(ctx context.Context) (err error) { if err := attachTheChain(ctx, inv, server); err != nil { return err } + bus, isNATS := server.Bus().(link.OverNATS) + if !isNATS { + return errors.New("the mesh's verbs are served over the bus, and this control plane is not on it") + } + // The condition store, opened before anything that raises or reads a condition starts beside the serving, so none + // of them meets conditionsFrom being written (issue 1945); closed after they all ended. + keeper := openServingConditions(ctx, open, bus) + if keeper != nil { + defer closeConditions(keeper) + } // The bus's own objects, asserted on every start. **Not created once at genesis**: a stream // somebody deleted, a mesh raised from a restored backup, or a bus whose data directory was @@ -176,7 +194,7 @@ func serve(ctx context.Context) (err error) { // And the pending assignments settled on a tick of their own (novox/hq ADR 0261): made once their module // is registered, ended with why when its build will not register it, raised and cleared as conditions. // Never by a read. - go settlingPending(ctx, open) + goServing(ctx, func() { settlingPending(ctx, open) }) // The durations the core's bounds are set from are kept a month (novox/hq to-be 45 Phase 0). // And what the catalogue decided a build meant. The builder's own result is already handled // above; this is the other half — the control plane is the only one of the three that knows @@ -224,10 +242,6 @@ func serve(ctx context.Context) (err error) { "Those answer the reason when called; everything else is served as usual\n", catalogue.ControllerSeatName, strings.Join(behind, ", ")) } - bus, isNATS := server.Bus().(link.OverNATS) - if !isNATS { - return errors.New("the mesh's verbs are served over the bus, and this control plane is not on it") - } // The hand-act log is counted for `status` on this connection rather than a new one a minute; no longer serving, the // connection is closed, and a verb run after in this process dials its own (a served test's leftover failed the // next test's `conditions` with "connection closed", issue 1945). @@ -247,18 +261,18 @@ func serve(ctx context.Context) (err error) { return err } // Composed now and kept current, before the verb that answers from it is served. - go statusFrom.keep(ctx) + goServing(ctx, func() { statusFrom.keep(ctx) }) // The facts snapshot a merge check is fed (novox/hq to-be 45 §9): kept current while this // controller holds the lease, read by the build seat from the artifact store. - go exportingFacts(ctx, open, bus.Conn.ConnectedServerVersion) + goServing(ctx, func() { exportingFacts(ctx, open, bus.Conn.ConnectedServerVersion) }) // The record mirrors (novox/hq ADR 0289, design 57 F3): the assignments and each registered module's // declared config on the bus, written from the store only when a record changed, by the lease holder. // The provisioner's plan of each handed module, as its machine was last sent it (design 57 P6), in the module's // assignment. inv.SetProvisioningPlans(inv.SentProvisioningPlans) - go mirroringRecords(ctx, open, inv, bus.Conn, bus) + goServing(ctx, func() { mirroringRecords(ctx, open, inv, bus.Conn, bus) }) // And the bucket of the files that travel by digest, judged against what each machine's latest declaration names (ADR 0486 §3). - go sweepingDeclarationFiles(ctx, inv, bus.Conn) + goServing(ctx, func() { sweepingDeclarationFiles(ctx, inv, bus.Conn) }) // And the modules the catalogue does not hold yet, registered from the builder's first done key of each, woken by // the builder's and the indexer's writes (novox/hq ADR 0501 §6, issue 1851): nothing asks a first build. watchNewModules(ctx, open) @@ -290,7 +304,7 @@ func serve(ctx context.Context) (err error) { // signals table, what the bus says about itself, and the self-check. A store that cannot be opened // is said and the controller serves on: status then says the conditions cannot be read, and is // not well — louder than not serving at all, and the push that repairs the bus still runs. - if stopWatching := watchTheMesh(ctx, open, server, bus); stopWatching != nil { + if stopWatching := watchTheMesh(ctx, open, server, bus, keeper); stopWatching != nil { defer stopWatching() } stopServing, err := bus.ServeSeatTools(catalogue.ControllerSeatName, handlers, log.New(os.Stdout, "", log.LstdFlags)) @@ -305,7 +319,7 @@ func serve(ctx context.Context) (err error) { } defer stopCLI() // Its read calls write no line each; how many each caller made is said once an hour (novox/hq issue 762). - go sayReadCalls(ctx) + goServing(ctx, func() { sayReadCalls(ctx) }) // And says so on the bus (novox/hq ADR 0197): what it serves, as the NATS services protocol asks. stopAnnouncing, err := bus.Announce(seatAnnouncement(handlers), log.New(os.Stdout, "", log.LstdFlags)) if err != nil { @@ -313,7 +327,10 @@ func serve(ctx context.Context) (err error) { } defer stopAnnouncing() - return server.Serve(ctx) + err = server.Serve(ctx) + // Ended and waited for before the verbs stop being served and the globals are put back, not after. + stopRunning() + return err } // OverlayCIDRVar is the range the mesh allocates node addresses from. diff --git a/cmd/mesh-controller/record_mirrors.go b/cmd/mesh-controller/record_mirrors.go index f2cdd90c..2f3853fd 100644 --- a/cmd/mesh-controller/record_mirrors.go +++ b/cmd/mesh-controller/record_mirrors.go @@ -399,14 +399,16 @@ func keepingDeclaredPackages(ctx context.Context, open *stores, api jetstream.Je } k := newPackagesKeeper(read, declaredPackagesFrom, sync, holds) for _, bucket := range []string{sdkdelivery.ControllerComposedBucket, broker.AssignmentsBucket} { - go wakeOnBucket(ctx, api, bucket, k.poke) + goServing(ctx, func() { wakeOnBucket(ctx, api, bucket, k.poke) }) } - go k.run(ctx, packagesTiming{safety: packagesSafetyEvery, gap: packagesGap}, func(why string) { - if why != "" { - fmt.Printf("the %s mirror cannot be kept: %s\n", m.Name, why) - } else { - fmt.Printf("the %s mirror is kept again\n", m.Name) - } + goServing(ctx, func() { + k.run(ctx, packagesTiming{safety: packagesSafetyEvery, gap: packagesGap}, func(why string) { + if why != "" { + fmt.Printf("the %s mirror cannot be kept: %s\n", m.Name, why) + } else { + fmt.Printf("the %s mirror is kept again\n", m.Name) + } + }) }) return k } diff --git a/cmd/mesh-controller/serving_group.go b/cmd/mesh-controller/serving_group.go new file mode 100644 index 00000000..cf054fa6 --- /dev/null +++ b/cmd/mesh-controller/serving_group.go @@ -0,0 +1,59 @@ +package main + +import ( + "context" + "sync" +) + +// **Every goroutine the serving controller starts ends before serve returns** (issue 1945). A loop or a watch started +// beside the serving runs until serve's context ends, and serve's stop waits for it before it closes the stores, the +// bus and the condition keeper those loops read: a loop that outlived its serve met the next serve's globals, its +// closed stores and its bus, and in the test process that is the next test (a race on conditionsFrom, a package that +// passed every test and still exited failing). Where it is checked: the tests that run serve whole, under -race. + +// servingGroup counts the goroutines one serve started. Once its stop began it starts nothing more: what would start is +// work for a controller that is stopping. +type servingGroup struct { + mu sync.Mutex + stopped bool + wg sync.WaitGroup +} + +type servingGroupKey struct{} + +// withServingGroup is ctx carrying a new group, which goServing under it counts. +func withServingGroup(ctx context.Context) (context.Context, *servingGroup) { + g := &servingGroup{} + return context.WithValue(ctx, servingGroupKey{}, g), g +} + +// 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()) { + g, _ := ctx.Value(servingGroupKey{}).(*servingGroup) + if g == nil { + go work() + return + } + g.mu.Lock() + if g.stopped { + g.mu.Unlock() + return + } + g.wg.Add(1) + g.mu.Unlock() + go func() { + defer g.wg.Done() + work() + }() +} + +// stop starts nothing more and waits for what was started; the caller ends their context first. A second stop waits +// for nothing. +func (g *servingGroup) stop() { + g.mu.Lock() + g.stopped = true + g.mu.Unlock() + g.wg.Wait() +} diff --git a/cmd/mesh-controller/states_announced.go b/cmd/mesh-controller/states_announced.go index d399528e..f68cdb0e 100644 --- a/cmd/mesh-controller/states_announced.go +++ b/cmd/mesh-controller/states_announced.go @@ -73,10 +73,10 @@ func keepAnnouncements(ctx context.Context, js *broker.JetStream, open *stores) if err != nil { return fmt.Errorf("following the states announced: %w", err) } - go func() { + goServing(ctx, func() { <-ctx.Done() consuming.Stop() - }() + }) return nil } diff --git a/cmd/mesh-controller/status_summary.go b/cmd/mesh-controller/status_summary.go index 3ea128f2..cc7b3a50 100644 --- a/cmd/mesh-controller/status_summary.go +++ b/cmd/mesh-controller/status_summary.go @@ -210,7 +210,7 @@ func (l nudgingListener) Heard(ctx context.Context, report link.Report) (bool, e // And, once the move of the secrets into the vault has started, what is left of this machine's (design 57's V3, // issue 676): a machine asleep when it started is moved when it is next heard from. Off the report's path. if err == nil && l.open != nil && report.Superseded == "" && report.Rekey == nil && report.Node != "" { - go moveWhenHeard(context.WithoutCancel(ctx), l.open, moveHeardOn(), report.Node) + goServing(ctx, func() { moveWhenHeard(context.WithoutCancel(ctx), l.open, moveHeardOn(), report.Node) }) } return news, err } diff --git a/cmd/mesh-controller/watchdogs.go b/cmd/mesh-controller/watchdogs.go index af6bc530..062b5b9f 100644 --- a/cmd/mesh-controller/watchdogs.go +++ b/cmd/mesh-controller/watchdogs.go @@ -545,10 +545,9 @@ func later(a, b time.Time) time.Time { return b } -// watchTheMesh opens the condition store and starts the watchdogs, the bus's advisories and the -// self-check, for the serving controller; the returned function stops them. Nil when the store could -// not be opened, which is said. -func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus link.OverNATS) func() { +// openServingConditions opens the condition store for the serving controller and makes it conditionsFrom. Nil when the store +// could not be opened, which is said: status then says the conditions cannot be read. +func openServingConditions(ctx context.Context, open *stores, bus link.OverNATS) *conditions.Keeper { keeper, err := keeperOn(ctx, bus.Conn, storeLooks{open.inventory}) if err != nil { fmt.Printf("the condition store could NOT be opened, so nothing that goes wrong is kept or said, and "+ @@ -556,6 +555,22 @@ func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus li return nil } conditionsFrom = keeper + return keeper +} + +// closeConditions flushes what the keeper holds and closes it, once nothing that raises a condition runs. +func closeConditions(keeper *conditions.Keeper) { + flushing, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + keeper.Close(flushing) +} + +// watchTheMesh starts the watchdogs, the bus's advisories and the self-check over the condition store's keeper, for the +// serving controller; the returned function stops them and waits for them to end. Nil when there is no keeper. +func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus link.OverNATS, keeper *conditions.Keeper) func() { + if keeper == nil { + return nil + } logf := func(format string, args ...any) { fmt.Printf(format+"\n", args...) } stopHearing, err := server.HearAdvisories(logf) if err != nil { @@ -568,26 +583,26 @@ func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus li d := &doctor{open: open, js: server.JetStream(), keeper: keeper, teller: bus, watchdogs: w, host: host} w.doctor = d doctorFrom = d + // Its own group as well as serve's: its stop waits for what it started. watching, stop := context.WithCancel(ctx) - go w.keep(watching) - go d.keep(watching) + watching, running := withServingGroup(watching) + goServing(watching, func() { w.keep(watching) }) + goServing(watching, func() { d.keep(watching) }) // And the healers (novox/hq to-be 45 §7): what is open that a registered healer answers, repaired // under the lease and the brake, every act said. healers := newHealing(open, keeper, bus, server.JetStream()) - go healers.keep(watching) + goServing(watching, func() { healers.keep(watching) }) // And the asker (novox/hq ADR 0259): what needs the operator and names its answers is asked of them, // and the answer chosen is performed on its warrant. startAsking(watching, open, server, bus.Conn, keeper) - go forgettingOldHeals(watching, open.inventory) + goServing(watching, func() { forgettingOldHeals(watching, open.inventory) }) fmt.Printf("watching the mesh: %d signal(s) every %s, %d probe(s) every %s; what is wrong is kept in %s "+ "and said as %s events\n", len(watchedRows()), watchEvery, len(runnableProbes()), doctorEvery, broker.ConditionsBucket, conditions.Seat) return func() { stop() + running.stop() stopHearing() - flushing, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - keeper.Close(flushing) } }