The node composes its own modules once the catalogue places it: one composer per node drains a mark set by any input's change, the agent applies what it composed when it differs, and a record move waits for a yes (issue 1126, ADR 0547 §2, §3)

2026-10-26 18:44:58 +01:00
parent 7fc0ea5478
commit 02d9d7b053
9 changed files with 866 additions and 5 deletions
@@ -0,0 +1,494 @@
+// Package selfcompose is the node agent's own composer (novox/hq ADR 0547 §2, issue 1126): it composes this node's
+// modules and its own composition with mesh-sdk's composer, from what is on the bus, and hands the compositions to the
+// agent as the agent once read them from the controller's bucket `composed-<node>`.
+//
+// **It watches, it never polls** (ADR 0547 §2): its inputs are watches of the node's wants, the indexer's placement and
+// modules, every node's own have key (where a node's record rides, ADR 0555 §3) and mesh-config's resolved values of this
+// node. A change of any of them marks the node to compose again; one loop drains the mark, so a burst of changes composes
+// the node once (composing one node takes 0.8 to 4.4 s, measured 2026-10-26). It hands the agent only the compositions that
+// differ from the ones it handed before, and takes back one no want names any more. A composition that cannot be made is
+// said, once per reason, and the ones handed before stand: a node keeps running what it ran rather than nothing.
+package selfcompose
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sort"
+	"strings"
+	"sync"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/compose/frombus"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+
+	"github.com/novox/mesh-host/internal/agent"
+)
+
+// ResolvedBucket is mesh-config's state of resolved values, keyed `<node>.<module>.<key>` (mesh-catalog modules/mesh-config
+// resolve.go ResolvedKey); a node reads its own slice alone (ADR 0547 §2).
+const ResolvedBucket = "mesh-config_resolved"
+
+// Env is what the composer needs of the node.
+type Env struct {
+	Node string
+	// Watch follows one bucket's keys matching pattern, as the agent's watches do.
+	Watch func(ctx context.Context, bucket, pattern string) (<-chan agent.Update, error)
+	// Get reads one key's latest value; nil and no error when there is none.
+	Get  func(ctx context.Context, bucket, key string) ([]byte, error)
+	Mesh frombus.Mesh
+	Say  func(string)
+}
+
+// Composer composes the node and is the agent's watcher of its compositions (agent.Env.Composed).
+type Composer struct {
+	env Env
+
+	mu         sync.Mutex
+	wants      map[string]delivery.Want // by module, the node's own under agent.NodeModule
+	placements map[string]delivery.Placement
+	records    map[string]delivery.NodeRecord
+	modules    map[string]delivery.Module
+	values     map[string]map[string]any // by module, this node's
+	manifests  map[string][]byte         // by digest, read once
+	handed     map[string]string         // composed key to the digest handed
+	said       string
+	caught     bool
+	rev        uint64
+}
+
+// New is a composer for one node.
+func New(env Env) *Composer {
+	if env.Say == nil {
+		env.Say = func(string) {}
+	}
+	return &Composer{env: env, wants: map[string]delivery.Want{}, placements: map[string]delivery.Placement{},
+		records: map[string]delivery.NodeRecord{}, modules: map[string]delivery.Module{}, values: map[string]map[string]any{},
+		manifests: map[string][]byte{}, handed: map[string]string{}}
+}
+
+// input is one watched bucket and how its updates are taken.
+type input struct {
+	bucket, pattern string
+	take            func(c *Composer, u agent.Update)
+}
+
+func (c *Composer) inputs() []input {
+	node := c.env.Node
+	return []input{
+		{delivery.AssignerWantsBucket, "want." + node + ".>", (*Composer).takeWant},
+		{delivery.PlacementBucket, ">", (*Composer).takePlacement},
+		{delivery.HaveBucket, "*." + delivery.HaveNodeWord, (*Composer).takeRecord},
+		{delivery.ModulesBucket, ">", (*Composer).takeModule},
+		{ResolvedBucket, node + ".>", (*Composer).takeValue},
+	}
+}
+
+// Watch starts the composer and answers the compositions it makes: each as a key of `composed-<node>` would be, then a
+// Caught update once every input was read and the node composed once, then every change. It ends when ctx ends or an
+// input's watch ends.
+func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, error) {
+	ins := c.inputs()
+	type tagged struct {
+		in int
+		u  agent.Update
+		ok bool
+	}
+	merged := make(chan tagged, 64)
+	for i, in := range ins {
+		ch, err := c.env.Watch(ctx, in.bucket, in.pattern)
+		if err != nil {
+			return nil, fmt.Errorf("the node composes from %s, which cannot be watched: %w", in.bucket, err)
+		}
+		go func(i int, ch <-chan agent.Update) {
+			for u := range ch {
+				select {
+				case merged <- tagged{in: i, u: u, ok: true}:
+				case <-ctx.Done():
+					return
+				}
+			}
+			select {
+			case merged <- tagged{in: i}:
+			case <-ctx.Done():
+			}
+		}(i, ch)
+	}
+	out := make(chan agent.Update, 64)
+	go func() {
+		defer close(out)
+		caught := make([]bool, len(ins))
+		ready := false
+		dirty := false
+		for {
+			// Drain whatever came, then compose once for all of it.
+			select {
+			case <-ctx.Done():
+				return
+			case t := <-merged:
+				if !t.ok {
+					c.env.Say(fmt.Sprintf("the watch of %s ended; the node composes no more until its agent starts again", ins[t.in].bucket))
+					return
+				}
+				if t.u.Caught {
+					caught[t.in] = true
+				} else {
+					ins[t.in].take(c, t.u)
+					dirty = true
+				}
+				if len(merged) > 0 {
+					continue
+				}
+			}
+			if !ready {
+				ready = all(caught)
+				if !ready {
+					continue
+				}
+				dirty = true
+			}
+			if !dirty {
+				continue
+			}
+			dirty = false
+			for _, u := range c.compose(ctx) {
+				select {
+				case out <- u:
+				case <-ctx.Done():
+					return
+				}
+			}
+			if !c.caughtSent() {
+				select {
+				case out <- agent.Update{Caught: true}:
+				case <-ctx.Done():
+					return
+				}
+			}
+		}
+	}()
+	return out, nil
+}
+
+func all(b []bool) bool {
+	for _, v := range b {
+		if !v {
+			return false
+		}
+	}
+	return true
+}
+
+// caughtSent says the Caught update went out, marking it sent the first time it is asked.
+func (c *Composer) caughtSent() bool {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	sent := c.caught
+	c.caught = true
+	return sent
+}
+
+func (c *Composer) takeWant(u agent.Update) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if _, err := delivery.ParseWantNodeKey(u.Key); err == nil {
+		if u.Deleted {
+			delete(c.wants, agent.NodeModule)
+		} else {
+			c.wants[agent.NodeModule] = delivery.Want{}
+		}
+		return
+	}
+	node, module, err := delivery.ParseWantKey(u.Key)
+	if err != nil || node != c.env.Node {
+		return
+	}
+	w, err := delivery.DecodeWant(u.Value)
+	if u.Deleted || err != nil || w.Removed || w.FP == "" {
+		delete(c.wants, module)
+		return
+	}
+	c.wants[module] = w
+}
+
+func (c *Composer) takePlacement(u agent.Update) {
+	node, err := delivery.ParsePlacementKey(u.Key)
+	if err != nil {
+		return
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	p, err := delivery.DecodePlacement(u.Value)
+	if u.Deleted || err != nil || p.Node != node {
+		delete(c.placements, node)
+		return
+	}
+	c.placements[node] = p
+}
+
+func (c *Composer) takeRecord(u agent.Update) {
+	node, _, isNode, err := delivery.ParseHaveKey(u.Key)
+	if err != nil || !isNode {
+		return
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	h, err := delivery.DecodeHaveNode(u.Value)
+	if u.Deleted || err != nil || h.Record == nil {
+		delete(c.records, node)
+		return
+	}
+	c.records[node] = *h.Record
+}
+
+func (c *Composer) takeModule(u agent.Update) {
+	name, err := delivery.ParseModulesKey(u.Key)
+	if err != nil {
+		return
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	m, err := delivery.DecodeModule(u.Value)
+	if u.Deleted || err != nil {
+		delete(c.modules, name)
+		return
+	}
+	c.modules[name] = m
+}
+
+func (c *Composer) takeValue(u agent.Update) {
+	rest, ok := strings.CutPrefix(u.Key, c.env.Node+".")
+	if !ok {
+		return
+	}
+	module, key, ok := strings.Cut(rest, ".")
+	if !ok {
+		return
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	var r struct {
+		Value json.RawMessage `json:"value"`
+	}
+	var v any
+	if u.Deleted || json.Unmarshal(u.Value, &r) != nil || len(r.Value) == 0 || json.Unmarshal(r.Value, &v) != nil {
+		delete(c.values[module], key)
+		return
+	}
+	if c.values[module] == nil {
+		c.values[module] = map[string]any{}
+	}
+	c.values[module][key] = v
+}
+
+// compose composes the node once and answers the updates to hand: each composition whose digest differs from the one
+// handed, and a deletion for each handed one no want names any more.
+func (c *Composer) compose(ctx context.Context) []agent.Update {
+	c.mu.Lock()
+	fps := map[string]string{}
+	kept := map[string]string{}
+	var behind []string // modules wanted at another build than the trunk's: composed with that build's manifest
+	for module, w := range c.wants {
+		if module == agent.NodeModule {
+			continue
+		}
+		fps[module] = w.FP
+		if m, ok := c.modules[module]; ok && m.FP2 != w.FP {
+			behind = append(behind, module)
+		}
+	}
+	_, nodeWanted := c.wants[agent.NodeModule]
+	r := reader{c: c}
+	c.mu.Unlock()
+	sort.Strings(behind)
+	for _, module := range behind {
+		digest, err := c.builtManifest(ctx, module, fps[module])
+		if err != nil {
+			c.sayOnce(fmt.Sprintf("%s is wanted at %s, whose manifest cannot be read, so the node does not compose: %v", module, fps[module], err))
+			return nil
+		}
+		kept[module] = digest
+	}
+
+	compositions, _, err := frombus.Compose(ctx, r, c.env.Mesh, frombus.Scope{Node: c.env.Node, Kept: kept}, fps)
+	if err != nil {
+		c.sayOnce(fmt.Sprintf("the node cannot compose its modules, and keeps running what it runs: %v", err))
+		return nil
+	}
+	c.sayOnce("")
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	var out []agent.Update
+	now := map[string]bool{}
+	keys := make([]string, 0, len(compositions))
+	for key := range compositions {
+		keys = append(keys, key)
+	}
+	sort.Strings(keys)
+	for _, key := range keys {
+		comp := compositions[key]
+		if comp.Module == "" && !nodeWanted {
+			continue
+		}
+		now[key] = true
+		dig, err := delivery.ComposedDigestOf(comp)
+		if err != nil {
+			c.env.Say(fmt.Sprintf("the composition %s has no digest, and is not handed: %v", key, err))
+			continue
+		}
+		if c.handed[key] == dig {
+			continue
+		}
+		raw, err := delivery.EncodeComposed(comp)
+		if err != nil {
+			c.env.Say(fmt.Sprintf("the composition %s cannot be written, and is not handed: %v", key, err))
+			continue
+		}
+		c.rev++
+		c.handed[key] = dig
+		out = append(out, agent.Update{Key: key, Value: raw, Revision: c.rev})
+	}
+	for key := range c.handed {
+		if !now[key] {
+			c.rev++
+			delete(c.handed, key)
+			out = append(out, agent.Update{Key: key, Deleted: true, Revision: c.rev})
+		}
+	}
+	return out
+}
+
+func (c *Composer) sayOnce(line string) {
+	c.mu.Lock()
+	if c.said == line {
+		c.mu.Unlock()
+		return
+	}
+	c.said = line
+	c.mu.Unlock()
+	if line != "" {
+		c.env.Say(line)
+	}
+}
+
+// builtManifest is the digest of the manifest a module's build at a fingerprint was made from (the builder's build key):
+// what a node composes a module from when its want names another build than the trunk's.
+func (c *Composer) builtManifest(ctx context.Context, module, fp string) (string, error) {
+	key, err := delivery.BuilderBuildKey(module, fp)
+	if err != nil {
+		return "", err
+	}
+	raw, err := c.env.Get(ctx, delivery.BuilderBuildsBucket, key)
+	if err != nil {
+		return "", err
+	}
+	if raw == nil {
+		return "", fmt.Errorf("the builder holds no build %s", key)
+	}
+	b, err := delivery.DecodeBuilderBuild(raw)
+	if err != nil {
+		return "", err
+	}
+	if b.Manifest == "" {
+		return "", fmt.Errorf("the build %s names no manifest", key)
+	}
+	return b.Manifest, nil
+}
+
+// Policy is a module's upgrade policy as its manifest at the indexer's trunk derives it (catalogue.DerivedUpgrade), for
+// the agent's judgement of a move (ADR 0547 §3); empty when the manifest is not known or cannot be read.
+func (c *Composer) Policy(module string) string {
+	c.mu.Lock()
+	m, ok := c.modules[module]
+	c.mu.Unlock()
+	if !ok || m.Manifest == "" {
+		return ""
+	}
+	raw, err := reader{c: c}.Manifest(context.Background(), m.Manifest)
+	if err != nil {
+		return ""
+	}
+	parsed, err := catalogue.ParseManifestForInputs(raw)
+	if err != nil {
+		return ""
+	}
+	policy, _, _ := catalogue.DerivedUpgrade(parsed)
+	return policy
+}
+
+// reader is the composer's view of the bus as frombus reads it.
+type reader struct{ c *Composer }
+
+func (r reader) Placements(context.Context) (map[string]delivery.Placement, error) {
+	r.c.mu.Lock()
+	defer r.c.mu.Unlock()
+	out := make(map[string]delivery.Placement, len(r.c.placements))
+	for k, v := range r.c.placements {
+		out[k] = v
+	}
+	return out, nil
+}
+
+func (r reader) Records(context.Context) (map[string]delivery.NodeRecord, error) {
+	r.c.mu.Lock()
+	defer r.c.mu.Unlock()
+	out := make(map[string]delivery.NodeRecord, len(r.c.records))
+	for k, v := range r.c.records {
+		out[k] = v
+	}
+	return out, nil
+}
+
+func (r reader) Modules(context.Context) (map[string]delivery.Module, error) {
+	r.c.mu.Lock()
+	defer r.c.mu.Unlock()
+	out := make(map[string]delivery.Module, len(r.c.modules))
+	for k, v := range r.c.modules {
+		out[k] = v
+	}
+	return out, nil
+}
+
+func (r reader) Manifest(ctx context.Context, digest string) ([]byte, error) {
+	r.c.mu.Lock()
+	raw, ok := r.c.manifests[digest]
+	r.c.mu.Unlock()
+	if ok {
+		return raw, nil
+	}
+	key, err := delivery.ManifestsKey(digest)
+	if err != nil {
+		return nil, err
+	}
+	value, err := r.c.env.Get(ctx, delivery.ManifestsBucket, key)
+	if err != nil {
+		return nil, err
+	}
+	if value == nil {
+		return nil, fmt.Errorf("the indexer holds no manifest %s", digest)
+	}
+	m, err := delivery.DecodeManifest(value)
+	if err != nil {
+		return nil, err
+	}
+	r.c.mu.Lock()
+	r.c.manifests[digest] = []byte(m.JSON)
+	r.c.mu.Unlock()
+	return []byte(m.JSON), nil
+}
+
+func (r reader) Values(_ context.Context, node, module string) (map[string]any, error) {
+	if node != r.c.env.Node {
+		return nil, nil // a node reads its own slice alone (ADR 0547 §2)
+	}
+	r.c.mu.Lock()
+	defer r.c.mu.Unlock()
+	if len(r.c.values[module]) == 0 {
+		return nil, nil
+	}
+	out := make(map[string]any, len(r.c.values[module]))
+	for k, v := range r.c.values[module] {
+		out[k] = v
+	}
+	return out, nil
+}
@@ -0,0 +1,159 @@
+package selfcompose
+
+import (
+	"context"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+
+	"github.com/novox/mesh-host/internal/agent"
+)
+
+const lemurs = `{"module":"lemurs","version":"1","capabilities":["systemd"],"resources":[{"id":"service","type":"service","unit":"lemurs.service","state":"running","boot":"enabled"}]}`
+
+// fakeBus is the buckets the composer watches, each a channel a test writes updates to, and the keys Get reads.
+type fakeBus struct {
+	mu    sync.Mutex
+	chans map[string]chan agent.Update
+	kv    map[string][]byte // bucket/key
+	said  []string
+}
+
+func newFakeBus() *fakeBus {
+	return &fakeBus{chans: map[string]chan agent.Update{}, kv: map[string][]byte{}}
+}
+
+func (b *fakeBus) ch(bucket string) chan agent.Update {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if b.chans[bucket] == nil {
+		b.chans[bucket] = make(chan agent.Update, 16)
+	}
+	return b.chans[bucket]
+}
+
+func (b *fakeBus) env() Env {
+	return Env{Node: "n1",
+		Watch: func(_ context.Context, bucket, _ string) (<-chan agent.Update, error) { return b.ch(bucket), nil },
+		Get: func(_ context.Context, bucket, key string) ([]byte, error) {
+			b.mu.Lock()
+			defer b.mu.Unlock()
+			return b.kv[bucket+"/"+key], nil
+		},
+		Say: func(s string) { b.mu.Lock(); b.said = append(b.said, s); b.mu.Unlock() }}
+}
+
+func (b *fakeBus) put(t *testing.T, bucket, key string, value []byte) {
+	t.Helper()
+	b.ch(bucket) <- agent.Update{Key: key, Value: value}
+}
+
+func next(t *testing.T, out <-chan agent.Update) agent.Update {
+	t.Helper()
+	select {
+	case u := <-out:
+		return u
+	case <-time.After(5 * time.Second):
+		t.Fatal("no update from the composer")
+		return agent.Update{}
+	}
+}
+
+func quiet(t *testing.T, out <-chan agent.Update) {
+	t.Helper()
+	select {
+	case u := <-out:
+		t.Fatalf("the composer handed %s, and nothing changed", u.Key)
+	case <-time.After(50 * time.Millisecond):
+	}
+}
+
+// Issue 1126 (ADR 0547 §2): the node composes its wanted modules and its own composition from its inputs on the bus once
+// every input was read, hands each, then Caught; an input that changes nothing composed hands nothing; a module no want
+// names any more is taken back; a node that cannot compose says so once and hands nothing.
+func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
+	b := newFakeBus()
+	fp := delivery.SourcePrefix2 + strings.Repeat("a", 64)
+	digest := delivery.ManifestDigest([]byte(lemurs))
+	man, _ := delivery.EncodeManifest(delivery.Manifest{Module: "lemurs", JSON: lemurs})
+	key, _ := delivery.ManifestsKey(digest)
+	b.kv[delivery.ManifestsBucket+"/"+key] = man
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	c := New(b.env())
+	out, err := c.Watch(ctx, "n1.>")
+	if err != nil {
+		t.Fatal(err)
+	}
+	place, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
+		Modules: []delivery.PlacedModule{{Module: "lemurs"}}})
+	have, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", Capabilities: []string{"systemd"}}})
+	mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"})
+	want, _ := delivery.EncodeWant(delivery.Want{FP: fp})
+	nodeWant, _ := delivery.EncodeNodeWant(delivery.NodeWant{})
+	b.put(t, delivery.PlacementBucket, "placement.n1", place)
+	b.put(t, delivery.HaveBucket, "n1.node", have)
+	b.put(t, delivery.ModulesBucket, "lemurs", mod)
+	b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", want)
+	b.put(t, delivery.AssignerWantsBucket, "want.n1.node", nodeWant)
+	for _, bucket := range []string{delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket} {
+		b.ch(bucket) <- agent.Update{Caught: true}
+	}
+	got := map[string]bool{}
+	for {
+		u := next(t, out)
+		if u.Caught {
+			break
+		}
+		c, err := delivery.DecodeComposed(u.Value)
+		if err != nil {
+			t.Fatalf("%s: %v", u.Key, err)
+		}
+		if c.Node != "n1" {
+			t.Errorf("%s composed for %q", u.Key, c.Node)
+		}
+		got[u.Key] = true
+	}
+	lemursKey, _ := delivery.ComposedKey("n1", "lemurs", fp)
+	nodeKey, _ := delivery.ComposedNodeKey("n1")
+	if !got[lemursKey] || !got[nodeKey] || len(got) != 2 {
+		b.mu.Lock()
+		t.Logf("said: %v", b.said)
+		b.mu.Unlock()
+		t.Fatalf("handed %v, want %s and %s", got, lemursKey, nodeKey)
+	}
+
+	// A config value of another module, and a value changing nothing composed, hand nothing.
+	b.put(t, ResolvedBucket, "n1.other.k", []byte(`{"value":1}`))
+	quiet(t, out)
+
+	// A want removed takes its composition back.
+	removed, _ := delivery.EncodeWant(delivery.Want{Removed: true})
+	b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", removed)
+	if u := next(t, out); u.Key != lemursKey || !u.Deleted {
+		t.Fatalf("a removed want handed %+v", u)
+	}
+
+	// A want of a module whose manifest cannot be read: said once, nothing handed, what was handed stands.
+	gone := delivery.SourcePrefix2 + strings.Repeat("b", 64)
+	missing, _ := delivery.EncodeWant(delivery.Want{FP: gone})
+	b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", missing)
+	quiet(t, out)
+	b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", missing)
+	quiet(t, out)
+	b.mu.Lock()
+	n := 0
+	for _, s := range b.said {
+		if strings.Contains(s, "lemurs is wanted at") {
+			n++
+		}
+	}
+	b.mu.Unlock()
+	if n != 1 {
+		t.Errorf("the fault was said %d times, want once: %v", n, b.said)
+	}
+}