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

@@ -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)
 	}

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

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)