From 2407bfb9ff3292543058e5bf9ee935b0f761fb2a Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 18:14:03 +0100 Subject: [PATCH 1/5] 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) -- 2.54.0 From 81e335e42dbf70b71bfcc093647d82783a655a29 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 18:15:22 +0100 Subject: [PATCH 2/5] Say why a node with no compositions answers none to compose again (ADR 0545, the empty-on-error rule) --- cmd/mesh-controller/kept_compositions.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cmd/mesh-controller/kept_compositions.go b/cmd/mesh-controller/kept_compositions.go index 148c5095..f7f188c1 100644 --- a/cmd/mesh-controller/kept_compositions.go +++ b/cmd/mesh-controller/kept_compositions.go @@ -101,7 +101,7 @@ func followKept(ctx context.Context, open *stores, node string, modules map[stri lister, err := at.bucket.ListKeys(ctx) if err != nil { if errors.Is(err, jetstream.ErrNoKeysFound) { - return nil, nil + return nil, nil // empty-on-error: a node with no compositions keeps none to compose again } return nil, err } -- 2.54.0 From e40b1e70db7a1d0dd15a3b49a832ae3c8187986f Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 18:27:43 +0100 Subject: [PATCH 3/5] Follow config from a durable consumer, compose again only what changed since, and hold the node only for a write (issue 1945, ADR 0545, review) The start composed every kept composition of every module on every node, each a whole node composed in memory under the node's lock: 10 to 50 minutes at a start, holding take-ins back. Now the writes of mesh-config_resolved are read from the durable consumer mesh-controller-config-follow, acknowledged once the change is composed, so a restart composes again only what was written since the last change handled, and a first start nothing it did not hear. The node's lock is taken for each write alone. A consumer that cannot be made or read raises config-not-followed and is tried again a minute later, cleared once it reads. The controller's account is granted that consumer's acknowledgements where mesh-config serves, as its verb is. Research 072: milestone 2, the composer reacting to the change of an input. --- cmd/mesh-controller/composed_start.go | 8 - cmd/mesh-controller/kept_compositions.go | 198 +++++++++++++++-------- cmd/mesh-controller/plain_words.go | 7 + internal/broker/composed.go | 4 + internal/broker/nats.go | 5 + 5 files changed, 151 insertions(+), 71 deletions(-) diff --git a/cmd/mesh-controller/composed_start.go b/cmd/mesh-controller/composed_start.go index 7f3fb165..698a0ff5 100644 --- a/cmd/mesh-controller/composed_start.go +++ b/cmd/mesh-controller/composed_start.go @@ -112,14 +112,6 @@ 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) } diff --git a/cmd/mesh-controller/kept_compositions.go b/cmd/mesh-controller/kept_compositions.go index f7f188c1..8258ff45 100644 --- a/cmd/mesh-controller/kept_compositions.go +++ b/cmd/mesh-controller/kept_compositions.go @@ -14,6 +14,7 @@ import ( "github.com/nats-io/nats.go/jetstream" "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/conditions" "github.com/novox/mesh-controller/internal/inventory" ) @@ -23,8 +24,9 @@ import ( // 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. +// inputs, writing only those whose composition differs (writeComposed). The writes are read from a durable consumer, +// acknowledged once handled, so a change made while no controller ran is delivered to the next one, and a start +// composes again only what was written since the last write handled, never every kept composition. // configTouched is the node and module a key of mesh-config_resolved (`..`) names. func configTouched(key string) (node, module string, ok bool) { @@ -92,8 +94,6 @@ func followKept(ctx context.Context, open *stores, node string, modules map[stri 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 @@ -160,7 +160,11 @@ func followKept(ctx context.Context, open *stores, node string, modules map[stri continue } key, _ := sdkdelivery.ComposedKey(node, k.module, k.fp) + // One write on the node's keys at a time in this process, held only for the write: composing in memory holds + // nothing, so a sweep never keeps a take-in from the node (the review of ADR 0545). + unlock := lockComposing(node) w, err := writeComposed(ctx, at, key, c) + unlock() if err != nil { errs = append(errs, err) continue @@ -175,84 +179,152 @@ func followKept(ctx context.Context, open *stores, node string, modules map[stri return wrote, errors.Join(errs...) } -// configFollowGather is how long the config watch waits after a write for more before it composes: mesh-config writes +// configFollowGather is how long the config consumer 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. +// configFollowConsumer is the controller's durable consumer of mesh-config's resolved values (ADR 0545 §1, §3). +const configFollowConsumer = broker.ConfigFollowConsumer + +// configFollowConfig is that consumer: each write from when it is made (a first start composes again nothing it did not +// hear), acknowledged after the write is handled, so what a stop cut is delivered to the next controller. +func configFollowConfig() jetstream.ConsumerConfig { + return jetstream.ConsumerConfig{Durable: configFollowConsumer, FilterSubject: "$KV." + broker.ConfigResolvedBucket + ".>", + DeliverPolicy: jetstream.DeliverNewPolicy, AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait, + MaxAckPending: takeInPending, Description: "the controller's following of config into kept compositions (novox/hq ADR 0545)"} +} + +// kindConfigNotFollowed is the condition of a config consumer that cannot be read: a change of config composes nothing +// until it is. +const kindConfigNotFollowed = "config-not-followed" + +func configNotFollowedObservation(err error) conditions.Observation { + return conditions.Observation{Scope: conditions.ScopeCore, ID: "config-follow", Token: kindConfigNotFollowed, + Kind: kindConfigNotFollowed, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator, Source: "config", + Summary: fmt.Sprintf("mesh-config's resolved values are not heard, so a change of config composes nothing: %v", err)} +} + +// watchConfigFollow reads the writes of mesh-config's resolved values from the controller's durable consumer and, for +// each node and module they touch, composes the node and the module's kept compositions there again, until ctx ends +// (ADR 0545). A consumer that cannot be made or read raises config-not-followed and is tried again a minute later; it is +// cleared once it is read. 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 { + err := followConfigOnce(ctx, open) + if ctx.Err() != nil { + return + } + fmt.Printf("config: %v; tried again in a minute\n", err) + if keeper := conditionsFrom; keeper != nil { + _, _ = keeper.Observe(ctx, configNotFollowedObservation(err)) + } 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 + case <-time.After(time.Minute): + } + } + }) +} + +// followConfigOnce reads the consumer until it fails or ctx ends. +func followConfigOnce(ctx context.Context, open *stores) error { + bus, err := aBus() + if err != nil { + return err + } + defer bus.Close() + js, err := jetstream.New(bus.Conn()) + if err != nil { + return err + } + c, err := js.CreateOrUpdateConsumer(ctx, "KV_"+broker.ConfigResolvedBucket, configFollowConfig()) + if err != nil { + return fmt.Errorf("the consumer of %s: %w", broker.ConfigResolvedBucket, err) + } + it, err := c.Messages(jetstream.PullMaxMessages(startTakeIns)) + if err != nil { + return fmt.Errorf("the consumer of %s: %w", broker.ConfigResolvedBucket, err) + } + defer it.Stop() + if keeper := conditionsFrom; keeper != nil { + _, _ = keeper.Clear(ctx, configNotFollowedObservation(nil).Key(), "mesh-config's resolved values are heard") + } + type next struct { + m jetstream.Msg + err error + } + got := make(chan next) + go func() { + for { + m, err := it.Next() + select { + case got <- next{m, err}: + case <-ctx.Done(): + if m != nil { + _ = m.Nak() } - if e == nil { - continue - } - node, module, ok := configTouched(e.Key()) - if !ok { - continue + return + } + if err != nil { + return + } + } + }() + var held []jetstream.Msg + touched := map[string]map[string]bool{} + var gather <-chan time.Time + for { + select { + case <-ctx.Done(): + for _, m := range held { + _ = m.Nak() // delivered at once to the next controller + } + return nil + case n := <-got: + if n.err != nil { + for _, m := range held { + _ = m.Nak() } + return fmt.Errorf("the consumer of %s stopped: %w", broker.ConfigResolvedBucket, n.err) + } + held = append(held, n.m) + key := strings.TrimPrefix(n.m.Subject(), "$KV."+broker.ConfigResolvedBucket+".") + if node, module, ok := configTouched(key); ok { if touched[node] == nil { touched[node] = map[string]bool{} } touched[node][module] = true - if gather == nil { - gather = time.After(configFollowGather) + } + if gather == nil { + gather = time.After(configFollowGather) + } + case <-gather: + gather = nil + now, msgs := touched, held + touched, held = map[string]map[string]bool{}, nil + forgetResolved() // the next composition reads the values just written, not a read of the last 5 s + failed := false + 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) + failed = true } - 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) - } + if _, err := followKept(ctx, open, node, modules); err != nil { + fmt.Printf("config: %s's kept compositions are not composed again: %v\n", node, err) + failed = true + } + } + for _, m := range msgs { + if failed { + _ = m.NakWithDelay(doneKeyRetry) // tried again: the bus delivers the writes again + } else { + _ = m.Ack() } } } - }) + } } // forgetResolved drops the last read of mesh-config's resolved values. diff --git a/cmd/mesh-controller/plain_words.go b/cmd/mesh-controller/plain_words.go index c1ebdf54..e8519748 100644 --- a/cmd/mesh-controller/plain_words.go +++ b/cmd/mesh-controller/plain_words.go @@ -84,6 +84,13 @@ var plainWordings = map[string]func(conditions.Observation) words{ "builds announced there wait until the controller starts again. The details say which and why.", Resolved: "the mesh reads where new builds are announced again"} }), + // Config not followed into the compositions (kept_compositions.go, novox/hq ADR 0545). + kindConfigNotFollowed: worded(func(o conditions.Observation) words { + return words{Headline: "Changes of settings are not reaching the machines", + Explanation: "The mesh could not read the settings as they change, so a changed setting is not prepared for " + + "the machines until it can. It tries again every minute; the details say why.", + Resolved: "changes of settings reach the machines again"} + }), // Machines and their node-engine. "silent": worded(func(o conditions.Observation) words { m := machineOr(o, "a machine") diff --git a/internal/broker/composed.go b/internal/broker/composed.go index b950e058..b7ff58a1 100644 --- a/internal/broker/composed.go +++ b/internal/broker/composed.go @@ -190,3 +190,7 @@ const TakeInConsumer = "mesh-controller-take-in" // TakeInBuckets are the buckets the take-in consumes: the indexer's modules (each module's trunk value) and the builder's // builds (each build's key). var TakeInBuckets = []string{delivery.ModulesBucket, delivery.BuilderBuildsBucket} + +// ConfigFollowConsumer is the durable consumer the controller follows mesh-config's resolved values with, on that +// bucket's stream (novox/hq ADR 0545). +const ConfigFollowConsumer = "mesh-controller-config-follow" diff --git a/internal/broker/nats.go b/internal/broker/nats.go index 734bffc1..66f66698 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -1273,6 +1273,11 @@ func PermissionsFor(p Principal) (Permissions, error) { pub = append(pub, "mesh.seat."+v.Seat+"."+CallKind+"."+v.Verb+"."+CallerToken(p.Username())) } } + // And its consumer of mesh-config's resolved values, acknowledged once a change is composed (novox/hq ADR 0545): + // where mesh-config serves, as its verb, so genesis grants nothing of mesh-config. + if len(p.AsksTheConfig) > 0 { + pub = append(pub, "$JS.ACK.KV_"+ConfigResolvedBucket+"."+ConfigFollowConsumer+".>") + } } sort.Strings(pub) -- 2.54.0 From 31a2bcdb3c76e05fdce38fccfc7a20fc6333fafd Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 18:38:45 +0100 Subject: [PATCH 4/5] Read each resolved config key once when the consumer is made, so values set before this release reach the kept compositions (issue 1945, ADR 0545, review 3) The config consumer delivered new writes only, and mesh-config never writes an unchanged value again: g14's mesh-build-go capacity of 2, set before this release, would never reach its kept compositions carrying 0. A newly made durable now reads each key's last value once, through the same gather, per-write lock and acknowledgement; its deliver policy is set before its first delivery. Acknowledgements are synchronous and sent past the stop, and one the bus refuses is said by its key. Research 072: milestone 2, the composer reacting to the change of an input. --- cmd/mesh-controller/composed_start_test.go | 6 ++++++ cmd/mesh-controller/kept_compositions.go | 16 ++++++++++++---- 2 files changed, 18 insertions(+), 4 deletions(-) diff --git a/cmd/mesh-controller/composed_start_test.go b/cmd/mesh-controller/composed_start_test.go index 0f988d51..d3b4054b 100644 --- a/cmd/mesh-controller/composed_start_test.go +++ b/cmd/mesh-controller/composed_start_test.go @@ -131,6 +131,12 @@ func TestAnEarlierFingerprintIsComposedFromItsOwnManifest(t *testing.T) { t.Errorf("%q touches %s %s (%v), want %v", k, node, module, ok, want) } } + // A value set before the consumer was made reaches the kept compositions: it reads each key once at creation. + if c := configFollowConfig(); c.DeliverPolicy != jetstream.DeliverLastPerSubjectPolicy || + c.AckPolicy != jetstream.AckExplicitPolicy { + t.Errorf("the config consumer delivers %v, acknowledged %v; want each key's last value, acknowledged", + c.DeliverPolicy, c.AckPolicy) + } }) } diff --git a/cmd/mesh-controller/kept_compositions.go b/cmd/mesh-controller/kept_compositions.go index 8258ff45..87aabfed 100644 --- a/cmd/mesh-controller/kept_compositions.go +++ b/cmd/mesh-controller/kept_compositions.go @@ -186,11 +186,15 @@ var configFollowGather = time.Second // configFollowConsumer is the controller's durable consumer of mesh-config's resolved values (ADR 0545 §1, §3). const configFollowConsumer = broker.ConfigFollowConsumer -// configFollowConfig is that consumer: each write from when it is made (a first start composes again nothing it did not -// hear), acknowledged after the write is handled, so what a stop cut is delivered to the next controller. +// configFollowConfig is that consumer: each resolved key's last value once when it is made, then each write, +// acknowledged after the write is handled, so what a stop cut is delivered to the next controller. The first start of +// the release that makes it composes again the kept compositions of every module with config, through the same gather, +// lock and acknowledgement: mesh-config never writes an unchanged value again, so a value set before this release (g14's +// mesh-build-go capacity of 2, against kept compositions carrying 0) would otherwise never reach them. Its deliver +// policy is set before its first delivery; an existing durable's cannot be changed. func configFollowConfig() jetstream.ConsumerConfig { return jetstream.ConsumerConfig{Durable: configFollowConsumer, FilterSubject: "$KV." + broker.ConfigResolvedBucket + ".>", - DeliverPolicy: jetstream.DeliverNewPolicy, AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait, + DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy, AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait, MaxAckPending: takeInPending, Description: "the controller's following of config into kept compositions (novox/hq ADR 0545)"} } @@ -320,7 +324,11 @@ func followConfigOnce(ctx context.Context, open *stores) error { if failed { _ = m.NakWithDelay(doneKeyRetry) // tried again: the bus delivers the writes again } else { - _ = m.Ack() + acking, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) + if err := m.DoubleAck(acking); err != nil { + fmt.Printf("config: the bus did not take the acknowledgement of %s: %v\n", m.Subject(), err) + } + cancel() } } } -- 2.54.0 From ff3cc47027e04e7d558259447fdf6f9620f9a552 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 18:43:38 +0100 Subject: [PATCH 5/5] Let the kept compositions' pass end at the stop (issue 1945, ADR 0545, review 3) followKept went on composing every kept key, and the config consumer every touched node, after the controller was told to stop. Both now check the stop before each key and each node; the writes of an unfinished batch are refused at once, so the next controller takes them. Research 072: milestone 2, the composer reacting to the change of an input. --- cmd/mesh-controller/kept_compositions.go | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/cmd/mesh-controller/kept_compositions.go b/cmd/mesh-controller/kept_compositions.go index 87aabfed..7dda518b 100644 --- a/cmd/mesh-controller/kept_compositions.go +++ b/cmd/mesh-controller/kept_compositions.go @@ -137,6 +137,9 @@ func followKept(ctx context.Context, open *stores, node string, modules map[stri var errs []error builds := map[string][]inventory.Build{} for _, k := range kept { + if ctx.Err() != nil { + return wrote, errors.Join(append(errs, ctx.Err())...) // stopped: the rest is the next controller's + } 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)) @@ -311,6 +314,10 @@ func followConfigOnce(ctx context.Context, open *stores) error { forgetResolved() // the next composition reads the values just written, not a read of the last 5 s failed := false for node, modules := range now { + if ctx.Err() != nil { + failed = true // stopped: the writes are refused below and delivered to the next controller + break + } 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) failed = true @@ -321,7 +328,9 @@ func followConfigOnce(ctx context.Context, open *stores) error { } } for _, m := range msgs { - if failed { + if failed && ctx.Err() != nil { + _ = m.Nak() // stopped: delivered at once to the next controller + } else if failed { _ = m.NakWithDelay(doneKeyRetry) // tried again: the bus delivers the writes again } else { acking, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) -- 2.54.0