End every goroutine serve starts before serve returns, so no loop outlives its controller (issue 1945)

The try of the on-change mirror failed with every test passed: goroutines a
served test started outlived its serve and met the next test's globals and its
closed stores. Under -race the served tests show it as a race on conditionsFrom
(watchTheMesh writing it while settlingPending, already running, read it), and
a take-in cut by the stop raised fingerprint-not-composed, a kind with no plain
words, which fails the package after its PASS line.

- serve counts every goroutine it and its parts start (serving_group.go) and
  ends and waits for them before anything else is undone: the lease's watch,
  the pending settler, the status summary, the facts, the mirrors and the new
  declared-packages keeper with its two bucket watches, the declaration sweep,
  the read-call counter, the watches' closers, the have watches, the vault move
  on a report, and the watchdogs, doctor, healers and asker under their own
  group.
- The condition store opens before anything that reads conditionsFrom starts;
  askerFrom is an atomic pointer, read from the keeper's goroutines.
- A done key's take-in cut by the stop raises nothing, and fingerprint-not-composed
  has plain words.

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

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.

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) })

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) {
 				})
 			}
 		}
-	}()
+	})
 }

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)
 	})
 }

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
 }

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}) })
 			}
 		}
-	}()
+	})
 }

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

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")

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.

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
 }

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()
+}

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
 }
 

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
 }

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)
 	}
 }