diff --git a/cmd/mesh-controller/composed_start.go b/cmd/mesh-controller/composed_start.go index 4db346e9..698a0ff5 100644 --- a/cmd/mesh-controller/composed_start.go +++ b/cmd/mesh-controller/composed_start.go @@ -316,20 +316,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..d3b4054b 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,97 @@ 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) + } + } + // 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) + } + }) } // 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..7dda518b --- /dev/null +++ b/cmd/mesh-controller/kept_compositions.go @@ -0,0 +1,352 @@ +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/conditions" + "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). 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) { + 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 + } + 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 // empty-on-error: a node with no compositions keeps none to compose again + } + 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 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)) + 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) + // 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 + } + 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 consumer waits after a write for more before it composes: mesh-config writes +// a module's keys one by one. +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 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.DeliverLastPerSubjectPolicy, 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) { + goServing(ctx, func() { + 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 <-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() + } + 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) + } + 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 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 + } + 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 && 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) + 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() + } + } + } + } +} + +// 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) 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)