Merge pull request 'Kept compositions follow config changes from a durable consumer, so going back to an earlier build keeps today's config (issue 1945, ADR 0545)' (#466) from fix/1945-kept-compositions-follow-config into main

Reviewed-on: #466

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
 }

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
+// An earlier fingerprint is composed from its own build's manifest with today's inputs: the controller this one
-// own fingerprint: the composition holds what that manifest declares, not what the controller registered now declares.
+// replaced (1945/AC6), and every kept composition of a module whose config changed (ADR 0545: on 2026-10-25 a want back
-func TestTheReplacedControllerIsComposedFromItsOwnManifest(t *testing.T) {
+// to an earlier mesh-build-go on g14 applied the capacity of 0 its kept composition was made with). The composition
-	now := catalogue.Manifest{Module: controllerModule, Version: "6", Resources: []map[string]any{
+// holds what that manifest declares and today's config, the kept keys are those of the touched modules at a
-		{"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings", "content": "six\n"}}}
+// fingerprint other than the registered one, the build of a fingerprint is its newest that worked, and a write of
-	before := catalogue.Manifest{Module: controllerModule, Version: "5", Resources: []map[string]any{
+// mesh-config's resolved values names the node and module it touches.
-		{"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings", "content": "five\n"}}}
+func TestAnEarlierFingerprintIsComposedFromItsOwnManifest(t *testing.T) {
-	in := compose.Input{Catalogue: map[string]catalogue.Manifest{controllerModule: now},
+	fp := func(c string) string { return sdkdelivery.SourcePrefix2 + strings.Repeat(c, 64) }
-		Nodes: []compose.Node{{Name: "anchor", Assigned: []string{controllerModule}, Stated: true}}}
+	t.Run("from its own manifest, with today's inputs", func(t *testing.T) {
-	fp := sdkdelivery.SourcePrefix2 + strings.Repeat("5", 64)
+		manifest := func(version string) catalogue.Manifest {
-	c, err := replacedComposition(in, replacedController{node: "anchor", fp: fp, manifest: before})
+			return catalogue.Manifest{Module: controllerModule, Version: version,
-	if err != nil {
+				Settings: map[string]catalogue.SettingDeclaration{
-		t.Fatal(err)
+					"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"}}}
 		}
-	if c.Module != controllerModule || c.FP != fp || c.Node != "anchor" || len(c.Resources) != 1 {
+		now, before := manifest("six"), manifest("five")
-		t.Fatalf("composed %s at %s on %s with %d resources", c.Module, c.FP, c.Node, len(c.Resources))
+		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\n" {
+			if f.Content != "five 4\n" {
-		t.Errorf("the replaced controller's composition holds %q, its own manifest says five", f.Content)
+				t.Errorf("%s holds %q, want its own manifest's five with today's capacity 4", name, f.Content)
-	}
-	if in.Catalogue[controllerModule].Version != "6" {
-		t.Errorf("composing the replaced controller changed the Input's catalogue")
 			}
 		}
+		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
+	t.Run("which build, which keys, which touch", func(t *testing.T) {
-// 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: "c6", SourceFingerprint: fp("6"), Manifest: m},
-		{ID: "c5-failed", SourceFingerprint: before, Failed: "no", Manifest: m},
+			{ID: "c5-failed", SourceFingerprint: fp("5"), Failed: "no", Manifest: m},
-		{ID: "c5", SourceFingerprint: before, Manifest: m},
+			{ID: "c5", SourceFingerprint: fp("5"), Manifest: m},
-		{ID: "c4", SourceFingerprint: older, Manifest: m},
+			{ID: "c4", SourceFingerprint: fp("4"), Manifest: m},
 		}
-	if b, ok := replacedBuild(builds, now); !ok || b.ID != "c5" {
+		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], now); 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

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 (`<node>.<module>.<key>`) 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()
+}

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)

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

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"

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)