From 2407bfb9ff3292543058e5bf9ee935b0f761fb2a Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 18:14:03 +0100 Subject: [PATCH] Compose kept compositions again when a module's config changes (issue 1945, ADR 0545) The controller composed a module only at the fingerprint it is registered at, so each of the up to five compositions kept for going back carried the config of the moment it was made: a want back to an earlier mesh-build-go on g14 put its capacity back to 0. Nothing composed a node when mesh-config's resolved values changed, either. Now a write of mesh-config_resolved (..), heard from that bucket's watch and gathered for a second, drops the controller's 5 s reading of the values, composes the node, and composes every kept fingerprint of the touched modules on that node again from its own build's manifest with today's inputs, writing only the compositions whose digest differs. A starting controller does it once for every node, for a change no controller heard. The two replaced-controller tests are one table test with the kept cases. Research 072: milestone 2, the composer reacting to the change of an input. --- cmd/mesh-controller/composed_start.go | 22 +- cmd/mesh-controller/composed_start_test.go | 129 ++++++---- cmd/mesh-controller/kept_compositions.go | 263 +++++++++++++++++++++ cmd/mesh-controller/new_modules.go | 3 + 4 files changed, 369 insertions(+), 48 deletions(-) create mode 100644 cmd/mesh-controller/kept_compositions.go diff --git a/cmd/mesh-controller/composed_start.go b/cmd/mesh-controller/composed_start.go index 4db346e9..7f3fb165 100644 --- a/cmd/mesh-controller/composed_start.go +++ b/cmd/mesh-controller/composed_start.go @@ -112,6 +112,14 @@ func composeAtStart(ctx context.Context, open *stores) { if err := composeReplacedController(ctx, open); err != nil { fmt.Printf("start: the controller this one replaced is not composed: %v\n", err) } + // And every kept composition with today's inputs, once: a change of config while no controller heard it (ADR 0545). + // One reading of the inputs for every node. + ctx = withInputCache(ctx) + for _, node := range names { + if _, err := followKept(ctx, open, node, nil); err != nil { + fmt.Printf("start: %s's kept compositions are not composed again: %v\n", node, err) + } + } if _, err := retireDeclarations(ctx, open); err != nil { fmt.Printf("start: whether the declarations' stream may go cannot be read: %v\n", err) } @@ -316,20 +324,26 @@ func composeReplacedController(ctx context.Context, open *stores) error { // replacedComposition is the controller's composition on r's node at r's fingerprint: in composed with r's manifest // in place of the controller's registered one. func replacedComposition(in compose.Input, r replacedController) (sdkdelivery.Composed, error) { + return keptComposition(in, r.node, controllerModule, r.fp, r.manifest) +} + +// keptComposition is module's composition on node at fp: in composed with manifest, that fingerprint's own, in place of +// the module's registered one, in the reading mode (nothing made). in is not changed. +func keptComposition(in compose.Input, node, module, fp string, manifest catalogue.Manifest) (sdkdelivery.Composed, error) { shelf := make(map[string]catalogue.Manifest, len(in.Catalogue)) for k, v := range in.Catalogue { shelf[k] = v } - shelf[controllerModule] = r.manifest + shelf[module] = manifest in.Catalogue = shelf - all, _, err := compose.Compositions(in, r.node, map[string]string{controllerModule: r.fp}, compose.Reading) + all, _, err := compose.Compositions(in, node, map[string]string{module: fp}, compose.Reading) if err != nil { return sdkdelivery.Composed{}, err } - key, _ := sdkdelivery.ComposedKey(r.node, controllerModule, r.fp) + key, _ := sdkdelivery.ComposedKey(node, module, fp) c, ok := all[key] if !ok { - return sdkdelivery.Composed{}, fmt.Errorf("%s composes no %s with the replaced build's manifest", r.node, controllerModule) + return sdkdelivery.Composed{}, fmt.Errorf("%s composes no %s with the manifest of %s", node, module, short(fp)) } return c, nil } diff --git a/cmd/mesh-controller/composed_start_test.go b/cmd/mesh-controller/composed_start_test.go index 83de4ab6..0f988d51 100644 --- a/cmd/mesh-controller/composed_start_test.go +++ b/cmd/mesh-controller/composed_start_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "slices" "strings" "testing" @@ -46,51 +47,91 @@ func TestTheFirstStartDeletesThePlansBucketsAndSaysEachOnce(t *testing.T) { } } -// 1945/AC6 (unit half): the controller this one replaced is composed on its node from its own build's manifest at its -// own fingerprint: the composition holds what that manifest declares, not what the controller registered now declares. -func TestTheReplacedControllerIsComposedFromItsOwnManifest(t *testing.T) { - now := catalogue.Manifest{Module: controllerModule, Version: "6", Resources: []map[string]any{ - {"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings", "content": "six\n"}}} - before := catalogue.Manifest{Module: controllerModule, Version: "5", Resources: []map[string]any{ - {"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings", "content": "five\n"}}} - in := compose.Input{Catalogue: map[string]catalogue.Manifest{controllerModule: now}, - Nodes: []compose.Node{{Name: "anchor", Assigned: []string{controllerModule}, Stated: true}}} - fp := sdkdelivery.SourcePrefix2 + strings.Repeat("5", 64) - c, err := replacedComposition(in, replacedController{node: "anchor", fp: fp, manifest: before}) - if err != nil { - t.Fatal(err) - } - if c.Module != controllerModule || c.FP != fp || c.Node != "anchor" || len(c.Resources) != 1 { - t.Fatalf("composed %s at %s on %s with %d resources", c.Module, c.FP, c.Node, len(c.Resources)) - } - var f struct{ Content string } - _ = json.Unmarshal(c.Resources[0], &f) - if f.Content != "five\n" { - t.Errorf("the replaced controller's composition holds %q, its own manifest says five", f.Content) - } - if in.Catalogue[controllerModule].Version != "6" { - t.Errorf("composing the replaced controller changed the Input's catalogue") - } -} +// An earlier fingerprint is composed from its own build's manifest with today's inputs: the controller this one +// replaced (1945/AC6), and every kept composition of a module whose config changed (ADR 0545: on 2026-10-25 a want back +// to an earlier mesh-build-go on g14 applied the capacity of 0 its kept composition was made with). The composition +// holds what that manifest declares and today's config, the kept keys are those of the touched modules at a +// fingerprint other than the registered one, the build of a fingerprint is its newest that worked, and a write of +// mesh-config's resolved values names the node and module it touches. +func TestAnEarlierFingerprintIsComposedFromItsOwnManifest(t *testing.T) { + fp := func(c string) string { return sdkdelivery.SourcePrefix2 + strings.Repeat(c, 64) } + t.Run("from its own manifest, with today's inputs", func(t *testing.T) { + manifest := func(version string) catalogue.Manifest { + return catalogue.Manifest{Module: controllerModule, Version: version, + Settings: map[string]catalogue.SettingDeclaration{ + "capacity": {Kind: catalogue.KindPreference, Default: float64(0), Why: "how many builds at once"}}, + Resources: []map[string]any{{"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings", + "content": version + " ${setting:capacity}\n"}}} + } + now, before := manifest("six"), manifest("five") + in := compose.Input{Catalogue: map[string]catalogue.Manifest{controllerModule: now}, + Nodes: []compose.Node{{Name: "anchor", Assigned: []string{controllerModule}, Stated: true, + Settings: map[string][]catalogue.Layer{controllerModule: {{From: "mesh", Values: map[string]any{"capacity": float64(4)}}}}}}} + for name, compose := range map[string]func() (sdkdelivery.Composed, error){ + "the replaced controller": func() (sdkdelivery.Composed, error) { + return replacedComposition(in, replacedController{node: "anchor", fp: fp("5"), manifest: before}) + }, + "a kept composition": func() (sdkdelivery.Composed, error) { + return keptComposition(in, "anchor", controllerModule, fp("5"), before) + }, + } { + c, err := compose() + if err != nil { + t.Fatalf("%s: %v", name, err) + } + if c.Module != controllerModule || c.FP != fp("5") || c.Node != "anchor" || len(c.Resources) != 1 { + t.Fatalf("%s: composed %s at %s on %s with %d resources", name, c.Module, c.FP, c.Node, len(c.Resources)) + } + var f struct{ Content string } + _ = json.Unmarshal(c.Resources[0], &f) + if f.Content != "five 4\n" { + t.Errorf("%s holds %q, want its own manifest's five with today's capacity 4", name, f.Content) + } + } + if in.Catalogue[controllerModule].Version != "six" { + t.Errorf("composing an earlier fingerprint changed the Input's catalogue") + } + }) -// 1945/AC6 (unit half): the controller replaced is the newest earlier build of the controller that worked and names a -// fingerprint of the new form, not the one registered now. -func TestTheReplacedControllerIsItsNewestEarlierBuild(t *testing.T) { - now, before, older := sdkdelivery.SourcePrefix2+strings.Repeat("6", 64), sdkdelivery.SourcePrefix2+strings.Repeat("5", 64), - sdkdelivery.SourcePrefix2+strings.Repeat("4", 64) - m := []byte(`{"module":"mesh-controller","version":"5"}`) - builds := []inventory.Build{ - {ID: "c6", SourceFingerprint: now, Manifest: m}, - {ID: "c5-failed", SourceFingerprint: before, Failed: "no", Manifest: m}, - {ID: "c5", SourceFingerprint: before, Manifest: m}, - {ID: "c4", SourceFingerprint: older, Manifest: m}, - } - if b, ok := replacedBuild(builds, now); !ok || b.ID != "c5" { - t.Errorf("the replaced build is %+v (%v), want c5", b, ok) - } - if _, ok := replacedBuild(builds[:1], now); ok { - t.Errorf("a controller with no earlier build replaced one") - } + t.Run("which build, which keys, which touch", func(t *testing.T) { + m := []byte(`{"module":"mesh-controller","version":"5"}`) + builds := []inventory.Build{ + {ID: "c6", SourceFingerprint: fp("6"), Manifest: m}, + {ID: "c5-failed", SourceFingerprint: fp("5"), Failed: "no", Manifest: m}, + {ID: "c5", SourceFingerprint: fp("5"), Manifest: m}, + {ID: "c4", SourceFingerprint: fp("4"), Manifest: m}, + } + if b, ok := replacedBuild(builds, fp("6")); !ok || b.ID != "c5" { + t.Errorf("the replaced build is %+v (%v), want c5", b, ok) + } + if _, ok := replacedBuild(builds[:1], fp("6")); ok { + t.Errorf("a controller with no earlier build replaced one") + } + if b, ok := buildAt(builds, fp("5")); !ok || b.ID != "c5" { + t.Errorf("the build at 5 is %+v (%v), want c5, the one that worked", b, ok) + } + if _, ok := buildAt(builds, fp("3")); ok { + t.Errorf("a fingerprint with no build has one") + } + key := func(node, module, f string) string { k, _ := sdkdelivery.ComposedKey(node, module, f); return k } + keys := []string{key("g14", "mesh-build-go", fp("2")), key("g14", "mesh-build-go", fp("1")), + key("g14", "mesh-build-go", fp("3")), key("g14", "hello-web", fp("1")), key("ace", "mesh-build-go", fp("1"))} + registered := map[string]string{"mesh-build-go": fp("3"), "hello-web": fp("9")} + got := keptToFollow(keys, "g14", map[string]bool{"mesh-build-go": true}, registered) + if want := []keptKey{{"mesh-build-go", fp("1")}, {"mesh-build-go", fp("2")}}; !slices.Equal(got, want) { + t.Errorf("kept to follow %v, want %v", got, want) + } + if got := keptToFollow(keys, "g14", nil, registered); len(got) != 3 { + t.Errorf("every module's kept compositions: %v", got) + } + for k, want := range map[string][2]string{"g14.mesh-build-go.capacity": {"g14", "mesh-build-go"}, + "g14.mesh-build-go.limits.cpu": {"g14", "mesh-build-go"}, "g14.mesh-build-go": {}, "": {}} { + node, module, ok := configTouched(k) + if [2]string{node, module} != want || ok != (want[0] != "") { + t.Errorf("%q touches %s %s (%v), want %v", k, node, module, ok, want) + } + } + }) } // 1945/AC14 (unit half): the declarations' stream stays while one node's have names another node-engine than the one diff --git a/cmd/mesh-controller/kept_compositions.go b/cmd/mesh-controller/kept_compositions.go new file mode 100644 index 00000000..148c5095 --- /dev/null +++ b/cmd/mesh-controller/kept_compositions.go @@ -0,0 +1,263 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/inventory" +) + +// **A kept composition follows its inputs** (novox/hq ADR 0545, issue 1945). The controller composes a module only at +// the fingerprint it is registered at, and keeps the compositions of up to five fingerprints (ADR 0539 ยง4) so that +// going back is one want. A kept one was never composed again, so it carried the config of the moment it was made: on +// 2026-10-25 a want back to an earlier mesh-build-go on g14 applied its capacity of 0, the value from before it was +// set. Now a write of mesh-config's resolved values for a module on a node, heard from that bucket, composes the node +// and then every kept fingerprint of that module on that node again, each from its own build's manifest with today's +// inputs, writing only those whose composition differs (writeComposed). A starting controller does it once for every +// node, for a change made while no controller heard it. + +// configTouched is the node and module a key of mesh-config_resolved (`..`) names. +func configTouched(key string) (node, module string, ok bool) { + parts := strings.SplitN(key, ".", 3) + if len(parts) != 3 || parts[0] == "" || parts[1] == "" { + return "", "", false + } + return parts[0], parts[1], true +} + +// keptKey is one kept composition to compose again. +type keptKey struct{ module, fp string } + +// keptToFollow is, of a node's composition keys, each of modules (every module when nil) at a fingerprint other than +// the one it is registered at, sorted: the kept compositions. The node's own composition and other nodes' are not. +func keptToFollow(keys []string, node string, modules map[string]bool, registered map[string]string) []keptKey { + var out []keptKey + for _, key := range keys { + n, module, fp, isNode, err := sdkdelivery.ParseComposedKey(key) + if err != nil || isNode || n != node || (modules != nil && !modules[module]) { + continue + } + if fp == registered[module] || !strings.HasPrefix(fp, sdkdelivery.SourcePrefix2) { + continue + } + out = append(out, keptKey{module, fp}) + } + sort.Slice(out, func(i, j int) bool { + if out[i].module != out[j].module { + return out[i].module < out[j].module + } + return out[i].fp < out[j].fp + }) + return out +} + +// buildAt is, of a module's builds newest first, the newest that worked at fp and kept its manifest. +func buildAt(builds []inventory.Build, fp string) (inventory.Build, bool) { + for _, b := range builds { + if b.Worked() && b.SourceFingerprint == fp && len(b.Manifest) > 0 { + return b, true + } + } + return inventory.Build{}, false +} + +// followKept composes again, on node, every kept composition of modules (every module when nil) with today's inputs, +// and writes those that differ. It answers the keys written; a fingerprint whose build's manifest is gone is said and +// left as it is. +func followKept(ctx context.Context, open *stores, node string, modules map[string]bool) ([]string, error) { + if _, err := theLease.epoch(ctx); err != nil { + return nil, err + } + bus, err := aBus() + if err != nil { + return nil, err + } + defer bus.Close() + js, err := jetstream.New(bus.Conn()) + if err != nil { + return nil, err + } + inv := open.inventory + registered, err := currentFingerprints(ctx, inv) + if err != nil { + return nil, err + } + unlock := lockComposing(node) + defer unlock() + at, err := openComposedPlace(ctx, js, bus.Conn(), node) + if err != nil { + return nil, err + } + lister, err := at.bucket.ListKeys(ctx) + if err != nil { + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + return nil, err + } + var keys []string + for k := range lister.Keys() { + keys = append(keys, k) + } + kept := keptToFollow(keys, node, modules, registered) + if len(kept) == 0 { + return nil, nil + } + sign, err := controllerSigner(ctx, open) + if err != nil { + return nil, err + } + upgrades, err := inv.Upgrades(ctx) + if err != nil { + return nil, err + } + at.sign = sign + at.policy = func(module string) (string, string, string) { + u, ok := upgrades[module] + if !ok { + return "", "", "" + } + return u.Policy(), u.From, u.Why + } + in, err := inputFor(ctx, open) + if err != nil { + return nil, err + } + var wrote []string + var errs []error + builds := map[string][]inventory.Build{} + for _, k := range kept { + if _, read := builds[k.module]; !read { + if builds[k.module], err = inv.Builds(ctx, k.module, 20); err != nil { + errs = append(errs, fmt.Errorf("%s's builds: %w", k.module, err)) + continue + } + } + b, found := buildAt(builds[k.module], k.fp) + if !found { + fmt.Printf("compositions: %s's kept %s on %s has no build whose manifest is kept, and is not composed again\n", + k.module, short(k.fp), node) + continue + } + var m catalogue.Manifest + if err := json.Unmarshal(b.Manifest, &m); err != nil { + errs = append(errs, fmt.Errorf("the manifest of %s's build %s: %w", k.module, b.ID, err)) + continue + } + c, err := keptComposition(in.Input, node, k.module, k.fp, m) + if err != nil { + errs = append(errs, fmt.Errorf("%s at %s on %s: %w", k.module, short(k.fp), node, err)) + continue + } + key, _ := sdkdelivery.ComposedKey(node, k.module, k.fp) + w, err := writeComposed(ctx, at, key, c) + if err != nil { + errs = append(errs, err) + continue + } + if w { + wrote = append(wrote, key) + } + } + if len(wrote) > 0 { + fmt.Printf("compositions: %s: kept compositions composed again with today's inputs: %s\n", node, strings.Join(wrote, ", ")) + } + return wrote, errors.Join(errs...) +} + +// configFollowGather is how long the config watch waits after a write for more before it composes: mesh-config writes +// a module's keys one by one. +var configFollowGather = time.Second + +// watchConfigFollow hears every write of mesh-config's resolved values and, for each node and module it touches, +// composes the node and the module's kept compositions there again, until ctx ends (ADR 0545). A bucket not there is +// said: a change is then composed by the next act that touches the node and the next start. +func watchConfigFollow(ctx context.Context, open *stores) { + bus, err := aBus() + if err != nil { + fmt.Printf("config: the resolved values are not heard: %v\n", err) + return + } + js, err := jetstream.New(bus.Conn()) + if err != nil { + bus.Close() + fmt.Printf("config: the resolved values are not heard: %v\n", err) + return + } + kv, err := js.KeyValue(ctx, broker.ConfigResolvedBucket) + if err != nil { + bus.Close() + fmt.Printf("config: %s is not heard, so a change of config composes nothing until the next act: %v\n", + broker.ConfigResolvedBucket, err) + return + } + w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly()) + if err != nil { + bus.Close() + fmt.Printf("config: %s is not heard: %v\n", broker.ConfigResolvedBucket, err) + return + } + goServing(ctx, func() { + defer bus.Close() + defer func() { _ = w.Stop() }() + touched := map[string]map[string]bool{} + var gather <-chan time.Time + for { + select { + case <-ctx.Done(): + return + case e, ok := <-w.Updates(): + if !ok { + if ctx.Err() == nil { + fmt.Printf("config: the watch of %s ended; a change is composed by the next start\n", broker.ConfigResolvedBucket) + } + return + } + if e == nil { + continue + } + node, module, ok := configTouched(e.Key()) + if !ok { + continue + } + if touched[node] == nil { + touched[node] = map[string]bool{} + } + touched[node][module] = true + if gather == nil { + gather = time.After(configFollowGather) + } + case <-gather: + gather = nil + now := touched + touched = map[string]map[string]bool{} + forgetResolved() // the next composition reads the values just written, not a read of the last 5 s + for node, modules := range now { + if err := composerFor(open).compose(ctx, []string{node}, false); err != nil { + fmt.Printf("config: %s is not composed after its config changed: %v\n", node, err) + } + if _, err := followKept(ctx, open, node, modules); err != nil { + fmt.Printf("config: %s's kept compositions are not composed again: %v\n", node, err) + } + } + } + } + }) +} + +// forgetResolved drops the last read of mesh-config's resolved values. +func forgetResolved() { + theResolved.mu.Lock() + theResolved.entries = nil + theResolved.mu.Unlock() +} diff --git a/cmd/mesh-controller/new_modules.go b/cmd/mesh-controller/new_modules.go index 7b014605..626a683b 100644 --- a/cmd/mesh-controller/new_modules.go +++ b/cmd/mesh-controller/new_modules.go @@ -208,6 +208,9 @@ func watchNewModules(ctx context.Context, open *stores) { // composed, the replaced controller composed, and from now on each new fingerprint registered, composed and granted. inBackground(ctx, func() { composeAtStart(ctx, open) }) watchDoneKeys(ctx, open) + // And a change of config composes the node and the touched modules' kept compositions again, heard from + // mesh-config's resolved values (ADR 0545, kept_compositions.go). + watchConfigFollow(ctx, open) // And each assignment that comes to stand issues its node's bus accounts and the module's grant (ADR 0543): a module // built long before is assigned to one more node with no build done again (issue 1945, assignment_accounts.go). watchAssignmentAccounts(ctx, open)