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..b03c07f4 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. @@ -80,7 +80,11 @@ func layBuildsOutSoon(ctx context.Context, inv *inventory.Inventory) { Source: "chain", Summary: "the builds are not laid out under their modules, so none is read: "} if err != nil { fmt.Printf("chain: the builds are not laid out under their modules, and none is read until they are: %v\n", err) - if keeper := conditionsFrom; keeper != nil && !failed { + if ctx.Err() != nil { + return // stopped: the next controller lays them out + } + // A cancelled layout is tried again and raises nothing: a cancellation is no fault of the chain (issue 1945). + if keeper := conditionsFrom; keeper != nil && !failed && !errors.Is(err, context.Canceled) { o.Summary += err.Error() if _, oerr := keeper.Observe(ctx, o); oerr == nil { failed = true diff --git a/cmd/mesh-controller/composed_keys.go b/cmd/mesh-controller/composed_keys.go index 8b8dd3ff..2b2ac625 100644 --- a/cmd/mesh-controller/composed_keys.go +++ b/cmd/mesh-controller/composed_keys.go @@ -166,10 +166,12 @@ func followDoneKey(ctx context.Context, open *stores, module string) { err := takeOnceInSlot(ctx, open, module) o := doneKeyObservation(module) if err == nil { - if keeper := conditionsFrom; keeper != nil && failed { + // Cleared on every success, whoever raised it: a controller before a restart or another follower (issue + // 1945). A condition that is not open costs one read. + if keeper := conditionsFrom; keeper != nil { _, _ = keeper.Clear(ctx, o.Key(), "composed and granted") - failed = false } + failed = false if _, again := doneKeysLookAgain.LoadAndDelete(module); again { continue // a wake came while this one ran: once more, for the key written meanwhile } @@ -194,7 +196,12 @@ func followDoneKey(ctx context.Context, open *stores, module string) { } continue } - if keeper := conditionsFrom; keeper != nil && !failed { + 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 && !errors.Is(err, context.Canceled) { + // A take-in whose own step was cancelled is tried again, and raises nothing: a cancellation is no fault + // of the build (issue 1945: 30 urgent conditions live, each a take-in cut by a restart). o.Summary += err.Error() if _, oerr := keeper.Observe(ctx, o); oerr == nil { failed = true @@ -302,18 +309,18 @@ 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) 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() { + inBackground(ctx, func() { doneKeysAtStart(ctx, catalogedModules(open), func(module string) { takeDoneKeyAtStart(ctx, open, module) }) }) } @@ -384,7 +391,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_keys_cancel_test.go b/cmd/mesh-controller/composed_keys_cancel_test.go new file mode 100644 index 00000000..0e34904c --- /dev/null +++ b/cmd/mesh-controller/composed_keys_cancel_test.go @@ -0,0 +1,83 @@ +package main + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/conditions" +) + +// A take-in cut by a cancellation raises no condition (issue 1945: on 2026-10-25 thirty urgent fingerprint-not-composed +// conditions were live, each a take-in a stop or a restart cut with "context canceled"): when the controller stops, the +// next one takes it in; when a step of its own was cancelled, it is tried again. +func TestATakeInCutByACancellationRaisesNothing(t *testing.T) { + for _, tc := range []struct { + name string + stop bool // the controller's context ends, else only the step's error is a cancellation + }{{"the controller stops", true}, {"a step was cancelled", false}} { + t.Run(tc.name, func(t *testing.T) { + was, wasRetry, wasSay, wasKeeper := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom + t.Cleanup(func() { + doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom = was, wasRetry, wasSay, wasKeeper + }) + store := conditions.NewInMemory() + k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, + Teller: &conditions.Told{}}) + t.Cleanup(func() { k.Close(context.Background()) }) + conditionsFrom = k + doneKeyRetry = time.Millisecond + doneKeysSay = func(string) {} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + calls := 0 + doneKeyTakeOnce = func(context.Context, *stores, string) error { + calls++ + if tc.stop { + cancel() // stopped mid-way + return fmt.Errorf("the module cannot be composed: %w", context.Canceled) + } + if calls == 1 { + return fmt.Errorf("the module cannot be composed: %w", context.Canceled) + } + return nil + } + done := make(chan struct{}) + go func() { takeDoneKey(ctx, nil, "cut-module"); close(done) }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("the take-in did not end") + } + if !tc.stop && calls != 2 { + t.Errorf("a take-in whose step was cancelled was tried %d time(s), want 2", calls) + } + if c, found, err := conditions.ReadOne(t.Context(), store, doneKeyObservation("cut-module").Key()); err != nil || found { + t.Fatalf("a take-in cut by a cancellation raised %+v (found %v, %v)", c, found, err) + } + }) + } +} + +// A take-in that succeeds clears the module's fingerprint-not-composed condition whoever raised it: a controller before a +// restart, or another follower (issue 1945: the operator read "a build needs a look" for take-ins that had since worked). +func TestASucceedingTakeInClearsAConditionAnotherRaised(t *testing.T) { + was, wasRetry, wasSay, wasKeeper := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom + t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom = was, wasRetry, wasSay, wasKeeper }) + store := conditions.NewInMemory() + k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Teller: &conditions.Told{}}) + t.Cleanup(func() { k.Close(context.Background()) }) + conditionsFrom = k + doneKeysSay = func(string) {} + o := doneKeyObservation("raised-before") + o.Summary += "the store refused the connection" + if _, err := k.Observe(t.Context(), o); err != nil { + t.Fatal(err) + } + doneKeyTakeOnce = func(context.Context, *stores, string) error { return nil } + takeDoneKey(t.Context(), nil, "raised-before") + if c, found, err := conditions.ReadOne(t.Context(), store, o.Key()); err != nil || found { + t.Fatalf("a take-in that succeeded left the condition another raised: %+v (found %v, %v)", c, found, err) + } +} diff --git a/cmd/mesh-controller/composed_rejudge.go b/cmd/mesh-controller/composed_rejudge.go index 0393123c..73215784 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 { @@ -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) } @@ -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/declaration_files.go b/cmd/mesh-controller/declaration_files.go index c87902ab..888e4eed 100644 --- a/cmd/mesh-controller/declaration_files.go +++ b/cmd/mesh-controller/declaration_files.go @@ -225,6 +225,9 @@ func filesNotPutObservation(node, summary string) conditions.Observation { // notePut records how putting a send's objects went: a failure counts, the second in a row raises the condition, a success // clears the count and the condition. func notePut(ctx context.Context, node string, put error) { + if errors.Is(put, context.Canceled) || ctx.Err() != nil { + return // a send cut by a cancellation or a stop says nothing of the bucket (issue 1945) + } filesFailMu.Lock() if put == nil { delete(filesFailed, node) diff --git a/cmd/mesh-controller/declaration_files_test.go b/cmd/mesh-controller/declaration_files_test.go index e96083ce..901d1a71 100644 --- a/cmd/mesh-controller/declaration_files_test.go +++ b/cmd/mesh-controller/declaration_files_test.go @@ -415,6 +415,7 @@ func TestAPutThatFailsSendAfterSendRaisesAConditionAndClearsWhenItWorks(t *testi was := conditionsFrom conditionsFrom = k t.Cleanup(func() { conditionsFrom = was }) + forgetPutFailures(t, "cond-node") key := filesNotPutObservation("cond-node", "").Key() open := func() bool { _, found, err := conditions.ReadOne(t.Context(), store, key) @@ -442,6 +443,36 @@ func TestAPutThatFailsSendAfterSendRaisesAConditionAndClearsWhenItWorks(t *testi } } +// A send cut by a cancellation counts as no failure to put its files, however often (issue 1945: a cancellation is no fault +// of the bucket). +func TestAPutCutByACancellationRaisesNothing(t *testing.T) { + store := conditions.NewInMemory() + k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Teller: &conditions.Told{}}) + t.Cleanup(func() { k.Close(context.Background()) }) + was := conditionsFrom + conditionsFrom = k + t.Cleanup(func() { conditionsFrom = was }) + forgetPutFailures(t, "cut-node") + cut := fmt.Errorf("the files could not be put: %w", context.Canceled) + for range filesNotPutAfter + 1 { + notePut(t.Context(), "cut-node", cut) + } + if _, found, err := conditions.ReadOne(t.Context(), store, filesNotPutObservation("cut-node", "").Key()); err != nil || found { + t.Fatalf("sends cut by a cancellation raised the condition (found %v, %v)", found, err) + } +} + +// forgetPutFailures starts the node's count of failed puts at nothing and leaves it so, as a run of the test again finds it. +func forgetPutFailures(t *testing.T, node string) { + forget := func() { + filesFailMu.Lock() + delete(filesFailed, node) + filesFailMu.Unlock() + } + forget() + t.Cleanup(forget) +} + // aBusThatSays is a bus that keeps the declarations published to it and, at each, asks whether the node's bucket is held. type aBusThatSays struct { link.Bus diff --git a/cmd/mesh-controller/declared_packages_integration_test.go b/cmd/mesh-controller/declared_packages_integration_test.go index c99ca3da..15a069d3 100644 --- a/cmd/mesh-controller/declared_packages_integration_test.go +++ b/cmd/mesh-controller/declared_packages_integration_test.go @@ -87,3 +87,29 @@ func TestTheDeclaredPackagesMirrorNamesWhatEachNodesModulesDeclare(t *testing.T) } } } + +// 1945/AC-mirror-5: the digest the declared-packages mirror skips an unchanged pass by is made from a real reading of the +// stores, is the same over two readings of the same records, and moves when an assignment does. +func TestTheInputDigestOfTheStoresIsStable(t *testing.T) { + open := aMesh(t) + ctx := t.Context() + register(t, open, catalogue.Manifest{Module: "viewer", Version: "1", Resources: []map[string]any{ + {"id": "pkg-curl", "type": "package", "package": "curl"}}}) + read := func() string { + in, err := readComposeInput(ctx, open) + if err != nil { + t.Fatal(err) + } + return inputDigest(in) + } + first, second := read(), read() + if first == "" || first != second { + t.Fatalf("two readings of the same records gave the digests %q and %q", first, second) + } + if _, err := assign(ctx, open, "anchor", "viewer"); err != nil { + t.Fatalf("assign: %v", err) + } + if read() == first { + t.Errorf("an assignment left the digest of the stores' Input at %s", first) + } +} diff --git a/cmd/mesh-controller/given.go b/cmd/mesh-controller/given.go index 5f240b57..1d905b01 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 { @@ -195,8 +195,8 @@ 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 0273ff3d..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. @@ -223,16 +223,16 @@ 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 // 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/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..70c96fc7 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,17 @@ 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) + defer stopRunning() // on an error return, the goroutines end before the keeper closes + } // 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 +195,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,19 +243,17 @@ 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). handActConn = bus.Conn defer func() { handActConn = nil }() + defer stopRunning() // on an error return, the goroutines end before handActConn is put back // And everything else this controller does on the bus for a moment (novox/hq issue 327). servingBus.Store(server.JetStream()) // No longer serving: nothing is lent, and dead-letters says it is not read here (novox/hq issue 330). defer servingBus.Store(nil) + defer stopRunning() // on an error return, the goroutines end before the bus is taken back // And says when it replaced a value given by hand (novox/hq ADR 0228). givenEvents = bus // And a pull request's merge check, asked when the forge announces its head and said when judged @@ -247,18 +264,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 +307,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 +322,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 +330,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 0f02cded..2f3853fd 100644 --- a/cmd/mesh-controller/record_mirrors.go +++ b/cmd/mesh-controller/record_mirrors.go @@ -2,12 +2,15 @@ package main import ( "context" + "crypto/sha256" + "encoding/hex" "encoding/json" "fmt" "sort" "time" "git.novox.be/novox/mesh-sdk/go/compose" + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -24,7 +27,8 @@ import ( // so no one call site sees every change; what every change has in common is that it is committed to the // store. So the serving controller reads the store every mirrorEvery and writes only what differs from what // the bucket holds (ADR 0289 §1): a quiet mesh writes nothing, and a change rolled back is never read. A -// mirror is therefore at most mirrorEvery behind its store. +// mirror is therefore at most mirrorEvery behind its store. The declared-packages mirror is the exception: it composes +// every node, so it passes on a change of what it composes from, not on this tick (keepingDeclaredPackages, issue 1945). // mirrorEvery is how often the mirrors are compared with the store: two small reads of the store. const mirrorEvery = 10 * time.Second @@ -45,21 +49,31 @@ func mirroringRecords(ctx context.Context, open *stores, inv *inventory.Inventor return e } sources := map[string]func(context.Context) (map[string][]byte, error){ - recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror, - recordmirror.Predefined.Name: inv.PredefinedMirror, - recordmirror.DeclaredPackages.Name: declaredPackagesMirror(open), + recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror, + recordmirror.Predefined.Name: inv.PredefinedMirror, } + packages := keepingDeclaredPackages(ctx, open, api, announce, epoch, holds) + held := false writers := map[string]*recordmirror.Writer{} failing := map[string]string{} tick := time.NewTicker(mirrorEvery) defer tick.Stop() events := &assignmentEvents{} for { - if holds() { + now := holds() + if now && !held { + // The lease taken: the declared-packages mirror is brought to the stores at once, not at its next change. + packages.poke() + } + held = now + if now { // The assignments are keys of the bus, written by the verbs that change them (novox/hq ADR 0408): this tick // keeps the provisioner's plan in their values and says what changed. They are no mirror of the store. keepingAssignments(ctx, inv, announce, epoch, events, failing) for _, m := range recordmirror.Mirrors() { + if sources[m.Name] == nil { + continue // the declared packages, kept on a change (keepingDeclaredPackages) + } why := mirrorOnce(ctx, api, m, writers, sources[m.Name], announce, epoch, holds) if why != failing[m.Name] { if why != "" { @@ -179,35 +193,263 @@ func keepingAssignments(ctx context.Context, inv *inventory.Inventory, announce events.known = now } -// declaredPackagesMirror is the source of the declared-packages mirror (ADR 0507 §5): each node's composition, asked as a -// question that makes nothing, and the packages it declares with the modules that declare them. A node whose own set does -// not compose has no key, and any other failure to read is the pass's error, so a key is never purged for a store that -// could not be read. +// declaredPackagesMirror is the source of the declared-packages mirror (ADR 0507 §5) over a fresh reading of the stores. func declaredPackagesMirror(open *stores) func(context.Context) (map[string][]byte, error) { return func(ctx context.Context) (map[string][]byte, error) { in, err := inputFor(ctx, open) if err != nil { return nil, err } - // Which module owns each resource is carried only for a node-engine that reads it, so every node is asked as one that does. - asked := in.Input - asked.Nodes = append([]compose.Node(nil), in.Nodes...) - for i := range asked.Nodes { - asked.Nodes[i].ReadsModules = true - } - out := map[string][]byte{} - for _, n := range in.sortedNodes() { - d, _, err := compose.ComposeIn(asked, n.Name, compose.Foreseeing) - if compose.IsUnresolvable(err) { - continue + return declaredPackagesFrom(ctx, in) + } +} + +// declaredPackagesFrom is the declared-packages mirror over an Input: each node's composition, asked as a question that makes +// nothing, and the packages it declares with the modules that declare them. A node whose own set does not compose has no key, +// and any other failure to read is the pass's error, so a key is never purged for a store that could not be read. +func declaredPackagesFrom(_ context.Context, in *meshInput) (map[string][]byte, error) { + // Which module owns each resource is carried only for a node-engine that reads it, so every node is asked as one that does. + asked := in.Input + asked.Nodes = append([]compose.Node(nil), in.Nodes...) + for i := range asked.Nodes { + asked.Nodes[i].ReadsModules = true + } + out := map[string][]byte{} + for _, n := range in.sortedNodes() { + d, _, err := compose.ComposeIn(asked, n.Name, compose.Foreseeing) + if compose.IsUnresolvable(err) { + continue + } + if err != nil { + return nil, fmt.Errorf("the packages %s declares cannot be read: %w", n.Name, err) + } + if out[n.Name], err = inventory.DeclaredPackagesValue(n.Name, d.Resources, d.ModuleOf); err != nil { + return nil, err + } + } + return out, nil +} + +// **The declared-packages mirror passes on a change, not on a tick** (issue 1945 note 22). Its source composes every node in +// full, about 45 s on the control node; on the 10 s tick it never rested and held the controller at 130-175 % CPU. Every +// change that can move a node's packages is a change of its composition, which the controller writes into the bucket of the +// compositions' digests, and of the assignments, which the verbs write into theirs: a watch of those two buckets wakes the +// keeper. A pass reads the stores' Input once and composes only when the Input's digest differs from the one it last +// mirrored, so a wake that changed nothing it composes from costs one reading. One pass at a time; the wakes during a pass +// are one pass after it, at least packagesGap after the last began. A rare safety pass (packagesSafetyEvery) covers a +// change no watched key says, and with an unchanged digest composes nothing either. +const ( + packagesSafetyEvery = 10 * time.Minute + packagesGap = 30 * time.Second +) + +// packagesTiming is when the declared-packages mirror passes: the safety pass's period and the least time between the +// starts of two passes. +type packagesTiming struct{ safety, gap time.Duration } + +// packagesKeeper keeps the declared-packages mirror: read is the Input, compose the mirror's values from it, sync writes +// those that differ from the bucket, holds says this controller holds the lease. Each a seam a test stands in for. +type packagesKeeper struct { + read func(ctx context.Context) (*meshInput, error) + compose func(ctx context.Context, in *meshInput) (map[string][]byte, error) + sync func(ctx context.Context, want map[string][]byte) error + holds func() bool + wake chan struct{} + // last is the digest of the Input last mirrored whole; empty after a failure or while the lease is not held, so the + // next pass composes. + last string +} + +func newPackagesKeeper(read func(context.Context) (*meshInput, error), + compose func(context.Context, *meshInput) (map[string][]byte, error), + sync func(context.Context, map[string][]byte) error, holds func() bool) *packagesKeeper { + return &packagesKeeper{read: read, compose: compose, sync: sync, holds: holds, wake: make(chan struct{}, 1)} +} + +// poke says something the mirror composes from may have changed; never blocks, and pokes before a pass are one. +func (k *packagesKeeper) poke() { + select { + case k.wake <- struct{}{}: + default: + } +} + +// pass brings the mirror to the Input, composing only when the Input differs from the one last mirrored. +func (k *packagesKeeper) pass(ctx context.Context) error { + if !k.holds() { + k.last = "" + return nil + } + in, err := k.read(ctx) + if err != nil { + k.last = "" + return fmt.Errorf("the store cannot be read: %w", err) + } + digest := inputDigest(in) + if digest != "" && digest == k.last { + return nil + } + want, err := k.compose(ctx, in) + if err != nil { + k.last = "" + return err + } + if err := k.sync(ctx, want); err != nil { + k.last = "" + return err + } + k.last = digest + return nil +} + +// run passes once at its start, then on each wake and each safety period, until ctx ends. said hears why a pass failed, +// and an empty line when one succeeds again. +func (k *packagesKeeper) run(ctx context.Context, at packagesTiming, said func(why string)) { + safety := time.NewTicker(at.safety) + defer safety.Stop() + failing := "" + var began time.Time + for { + // The wakes so far are this pass's: drained before it reads, so a change after is a pass of its own. + select { + case <-k.wake: + default: + } + began = time.Now() + why := "" + if err := k.pass(ctx); err != nil { + why = err.Error() + } + if why != failing { + said(why) + failing = why + } + select { + case <-ctx.Done(): + return + case <-k.wake: + case <-safety.C: + } + if rest := at.gap - time.Since(began); rest > 0 { + select { + case <-ctx.Done(): + return + case <-time.After(rest): } + } + } +} + +// inputDigest is a digest of everything the composition reads (the Input, and why a node or the bus is refused): the same +// for the same Input. Empty when it cannot be made, and then the pass composes. +func inputDigest(in *meshInput) string { + refused := map[string]string{} + for node, err := range in.refused { + if err != nil { + refused[node] = err.Error() + } + } + bus := "" + if in.busRefused != nil { + bus = in.busRefused.Error() + } + body, err := json.Marshal(struct { + Input compose.Input + Refused map[string]string + BusRefused string + }{in.Input, refused, bus}) + if err != nil { + return "" + } + sum := sha256.Sum256(body) + return hex.EncodeToString(sum[:]) +} + +// keepingDeclaredPackages keeps the declared-packages mirror for as long as ctx: woken by the watch of the compositions' +// digests and of the assignments. It answers the keeper, whose poke a caller may also wake it with. +func keepingDeclaredPackages(ctx context.Context, open *stores, api jetstream.JetStream, announce recordmirror.Announcer, + epoch func() uint64, holds func() bool) *packagesKeeper { + m := recordmirror.DeclaredPackages + var w *recordmirror.Writer + read := func(ctx context.Context) (*meshInput, error) { + reading, cancel := context.WithTimeout(ctx, 2*time.Minute) + defer cancel() + return readComposeInput(reading, open) + } + sync := func(ctx context.Context, want map[string][]byte) error { + writing, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + if w == nil { + kv, err := api.KeyValue(writing, m.Bucket) if err != nil { - return nil, fmt.Errorf("the packages %s declares cannot be read: %w", n.Name, err) + return fmt.Errorf("the bucket %s cannot be read: %v", m.Bucket, err) + } + w = &recordmirror.Writer{Mirror: m, KV: kv, Announce: announce, Epoch: epoch, Holds: holds} + } + lost := w.LostEvents + r, err := w.Sync(writing, want) + if r.Written+r.Removed > 0 { + fmt.Printf("the %s mirror moved: %d written, %d removed\n", m.Name, r.Written, r.Removed) + } + if w.LostEvents > lost { + fmt.Printf("the %s mirror lost %d event(s): the bucket holds the change, its notice was refused twice\n", + m.Name, w.LostEvents-lost) + } + return err + } + k := newPackagesKeeper(read, declaredPackagesFrom, sync, holds) + for _, bucket := range []string{sdkdelivery.ControllerComposedBucket, broker.AssignmentsBucket} { + goServing(ctx, func() { wakeOnBucket(ctx, api, bucket, k.poke) }) + } + 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 +} + +// wakeOnBucket calls wake on every write or delete of a key in the bucket, for as long as ctx: only the keys' metadata is +// watched. A bucket not there yet, or a watch that ends, is watched again a minute later; the safety pass covers the gap. +func wakeOnBucket(ctx context.Context, api jetstream.JetStream, bucket string, wake func()) { + for { + if err := watchBucketOnce(ctx, api, bucket, wake); err != nil && ctx.Err() == nil { + fmt.Printf("the declared-packages mirror does not hear %s: %v\n", bucket, err) + } + select { + case <-ctx.Done(): + return + case <-time.After(time.Minute): + } + } +} + +// watchBucketOnce watches the bucket until its watch ends or ctx does. +func watchBucketOnce(ctx context.Context, api jetstream.JetStream, bucket string, wake func()) error { + kv, err := api.KeyValue(ctx, bucket) + if err != nil { + return err + } + watch, err := kv.WatchAll(ctx, jetstream.UpdatesOnly(), jetstream.MetaOnly()) + if err != nil { + return err + } + defer func() { _ = watch.Stop() }() + for { + select { + case <-ctx.Done(): + return nil + case e, ok := <-watch.Updates(): + if !ok { + return fmt.Errorf("the watch ended") } - if out[n.Name], err = inventory.DeclaredPackagesValue(n.Name, d.Resources, d.ModuleOf); err != nil { - return nil, err + if e != nil { + wake() } } - return out, nil } } diff --git a/cmd/mesh-controller/record_mirrors_test.go b/cmd/mesh-controller/record_mirrors_test.go new file mode 100644 index 00000000..1d01bee2 --- /dev/null +++ b/cmd/mesh-controller/record_mirrors_test.go @@ -0,0 +1,191 @@ +package main + +import ( + "context" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/compose" +) + +// The declared-packages mirror composes only when what it composes from changed (issue 1945 note 22: it composed every node +// in full every 10 s, a pass of about 45 s, so the controller on the control node never rested at 130-175 % CPU). + +// scaled is a cadence of the mirror with a second taken as a millisecond, so a test's minute is 60 ms. +func scaled() packagesTiming { + return packagesTiming{safety: packagesSafetyEvery / 1000, gap: packagesGap / 1000} +} + +const scaledMinute = 60 * time.Millisecond + +// fakeMesh is the keeper's input as a test changes it: each node's assigned modules, and the compositions counted by node. +type fakeMesh struct { + mu sync.Mutex + assigned map[string][]string + composed atomic.Int64 // node compositions + synced atomic.Int64 + lastSync map[string][]byte + // block, when set, holds the next composition until it is closed; entered is closed when that composition began. + block chan struct{} + entered chan struct{} +} + +func newFakeMesh(nodes int) *fakeMesh { + f := &fakeMesh{assigned: map[string][]string{}} + for i := 0; i < nodes; i++ { + f.assigned[fmt.Sprintf("node%d", i)] = []string{"base"} + } + return f +} + +func (f *fakeMesh) assign(node, module string) { + f.mu.Lock() + defer f.mu.Unlock() + f.assigned[node] = append(append([]string(nil), f.assigned[node]...), module) +} + +func (f *fakeMesh) keeper() *packagesKeeper { + read := func(context.Context) (*meshInput, error) { + f.mu.Lock() + defer f.mu.Unlock() + in := compose.Input{Catalogue: map[string]catalogue.Manifest{"base": {Module: "base", Version: "1"}}} + for _, n := range nodesOfFake(f.assigned) { + in.Nodes = append(in.Nodes, compose.Node{Name: n, Assigned: append([]string(nil), f.assigned[n]...)}) + } + return &meshInput{Input: in}, nil + } + composeAll := func(_ context.Context, in *meshInput) (map[string][]byte, error) { + f.mu.Lock() + block, entered := f.block, f.entered + f.block = nil + f.mu.Unlock() + if block != nil { + close(entered) + <-block + } + out := map[string][]byte{} + for _, n := range in.Nodes { + f.composed.Add(1) + out[n.Name] = []byte(fmt.Sprint(n.Assigned)) + } + return out, nil + } + syncAll := func(_ context.Context, want map[string][]byte) error { + f.mu.Lock() + f.lastSync = want + f.mu.Unlock() + f.synced.Add(1) + return nil + } + return newPackagesKeeper(read, composeAll, syncAll, func() bool { return true }) +} + +func nodesOfFake(m map[string][]string) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + for i := range out { + for j := i + 1; j < len(out); j++ { + if out[j] < out[i] { + out[i], out[j] = out[j], out[i] + } + } + } + return out +} + +func runKeeper(t *testing.T, k *packagesKeeper) context.CancelFunc { + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + k.run(ctx, scaled(), func(string) {}) + }() + t.Cleanup(func() { + cancel() + <-done + }) + return cancel +} + +func eventually(t *testing.T, what string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(time.Second) + for !cond() { + if time.Now().After(deadline) { + t.Fatalf("not within a second: %s", what) + } + time.Sleep(time.Millisecond) + } +} + +// 1945/AC-mirror-1: given a mesh whose input does not change, when a minute passes after the mirror's first pass, then it +// composes no node (before: a pass every 10 s, each composing every node). The count per minute is said. +func TestTheDeclaredPackagesMirrorComposesNothingWithoutAChange(t *testing.T) { + const nodes = 5 + f := newFakeMesh(nodes) + runKeeper(t, f.keeper()) + eventually(t, "the first pass", func() bool { return f.synced.Load() >= 1 }) + start := f.composed.Load() + time.Sleep(scaledMinute) + perMinute := f.composed.Load() - start + t.Logf("node compositions in a minute of an unchanged mesh of %d nodes: %d", nodes, perMinute) + if perMinute != 0 { + t.Errorf("an unchanged mesh of %d nodes was composed %d times in a minute; want 0", nodes, perMinute) + } +} + +// 1945/AC-mirror-2: given a pass that is running, when fifty changes arrive during it, then exactly one more pass follows, +// never two at once and never one per change. +func TestTheDeclaredPackagesMirrorPassesOncePerBurst(t *testing.T) { + f := newFakeMesh(1) + release := make(chan struct{}) + f.block, f.entered = release, make(chan struct{}) + k := f.keeper() + runKeeper(t, k) + <-f.entered + for i := 0; i < 50; i++ { + f.assign("node0", fmt.Sprintf("m%d", i)) + k.poke() + } + close(release) + eventually(t, "the pass after the burst", func() bool { return f.synced.Load() >= 2 }) + time.Sleep(2*scaled().gap + 20*time.Millisecond) + if got := f.synced.Load(); got != 2 { + t.Errorf("a burst of 50 changes during a pass gave %d passes in all; want 2 (the running one and one after)", got) + } +} + +// 1945/AC-mirror-3: given the mirror at rest, when an input changes and its event arrives, then the change is mirrored. +func TestTheDeclaredPackagesMirrorMirrorsAChange(t *testing.T) { + f := newFakeMesh(2) + k := f.keeper() + runKeeper(t, k) + eventually(t, "the first pass", func() bool { return f.synced.Load() >= 1 }) + f.assign("node1", "viewer") + k.poke() + eventually(t, "the change mirrored", func() bool { + f.mu.Lock() + defer f.mu.Unlock() + return string(f.lastSync["node1"]) == "[base viewer]" + }) +} + +// 1945/AC-mirror-4: the digest of the input is the same for the same input and differs when an assignment differs. +func TestTheInputDigestFollowsTheInput(t *testing.T) { + of := func(assigned ...string) *meshInput { + return &meshInput{Input: compose.Input{Nodes: []compose.Node{{Name: "anchor", Assigned: assigned}}}} + } + a, b, c := inputDigest(of("base")), inputDigest(of("base")), inputDigest(of("base", "viewer")) + if a == "" || a != b { + t.Errorf("the same input gave the digests %q and %q", a, b) + } + if a == c { + t.Errorf("an added assignment left the digest at %q", a) + } +} diff --git a/cmd/mesh-controller/serving_group.go b/cmd/mesh-controller/serving_group.go new file mode 100644 index 00000000..f810be0c --- /dev/null +++ b/cmd/mesh-controller/serving_group.go @@ -0,0 +1,60 @@ +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. 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 true + } + g.mu.Lock() + if g.stopped { + g.mu.Unlock() + return false + } + g.wg.Add(1) + g.mu.Unlock() + go 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 +// 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..dbd17204 100644 --- a/cmd/mesh-controller/states_announced.go +++ b/cmd/mesh-controller/states_announced.go @@ -67,16 +67,16 @@ 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 { 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) } }