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 (<node>.<module>.<key>), 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.
2026-10-25 18:14:03 +01:00
parent 850f4b4d26
commit 2407bfb9ff
4 changed files with 369 additions and 48 deletions
@@ -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 (`<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
+	}
+	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()
+}