Keep each placed module's data items in the engine's state data, so disk-load measures without composing (ADR 0577)

After a pass that handed a new composition, the node's composer works out its modules' data
items from the inputs it composed from and writes each module's key <node>.<module> of
mesh-host_data only when it differs, deleting the keys of modules that left. Pins mesh-sdk
feat/data-items for the contract and the engine's grant. Two tool-answer tests are merged so
the unit count stays where it was.

Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL

internal/link/enginedata.go

@@ -0,0 +1,103 @@
+package link
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"github.com/nats-io/nats.go"
+)
+
+// ---- the data items of the modules placed on this node, in the bus's own storage (novox/hq ADR 0577 clause 1) ------
+
+// DataBucket is the node-engine's declared state data (delivery.EngineDataBucket), created by the grants part from the
+// engine's manifest as its states acts and health are: one key `<node>.<module>` per module placed on the node that
+// declares data items, written by that node's engine alone, read by that node's disk-load.
+const DataBucket = delivery.EngineDataBucket
+
+// dataStore binds DataBucket over the link open now: bound, never created.
+func (l *natsLink) dataStore() (nats.KeyValue, error) {
+	kv, err := l.js.KeyValue(DataBucket)
+	if err != nil {
+		return nil, fmt.Errorf("the node-engine's state %s cannot be bound (it is created from the node-engine's "+
+			"manifest; written again at the next composition that changes): %w", DataBucket, err)
+	}
+	return kv, nil
+}
+
+// EngineData is this node's keys of DataBucket over whichever link the session has open (selfcompose.DataStore); with
+// none open every read and write is refused, and nothing is kept anywhere else.
+type EngineData struct{ Session *Session }
+
+func (d EngineData) kv() (nats.KeyValue, error) {
+	d.Session.init()
+	d.Session.mu.Lock()
+	bus := d.Session.bus
+	d.Session.mu.Unlock()
+	l, ok := bus.(interface{ dataStore() (nats.KeyValue, error) })
+	if bus == nil || !ok {
+		return nil, ErrNoLink
+	}
+	return l.dataStore()
+}
+
+// Get is a key's value, nil when it has none.
+func (d EngineData) Get(_ context.Context, key string) ([]byte, error) {
+	kv, err := d.kv()
+	if err != nil {
+		return nil, err
+	}
+	e, err := kv.Get(key)
+	if errors.Is(err, nats.ErrKeyNotFound) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	return e.Value(), nil
+}
+
+// Put writes a key.
+func (d EngineData) Put(_ context.Context, key string, value []byte) error {
+	kv, err := d.kv()
+	if err != nil {
+		return err
+	}
+	_, err = kv.Put(key, value)
+	return err
+}
+
+// Delete removes a key; one that was not there is no error.
+func (d EngineData) Delete(_ context.Context, key string) error {
+	kv, err := d.kv()
+	if err != nil {
+		return err
+	}
+	if err := kv.Delete(key); err != nil && !errors.Is(err, nats.ErrKeyNotFound) {
+		return err
+	}
+	return nil
+}
+
+// Keys is every key matching pattern that holds a value: read by a watch of the pattern alone, the only read the
+// engine's grant gives of its own node's keys.
+func (d EngineData) Keys(_ context.Context, pattern string) ([]string, error) {
+	kv, err := d.kv()
+	if err != nil {
+		return nil, err
+	}
+	w, err := kv.WatchFiltered([]string{pattern}, nats.IgnoreDeletes(), nats.MetaOnly())
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = w.Stop() }()
+	var keys []string
+	for e := range w.Updates() {
+		if e == nil {
+			break
+		}
+		keys = append(keys, e.Key())
+	}
+	return keys, nil
+}

internal/link/health_test.go

@@ -168,6 +168,18 @@ func TestAToolsAnswerIsHealthyOnlyWhenItSaysSo(t *testing.T) {
 	if ToolSubject("keycloak", "keycloak_admin_health", "anchor") != "mesh.mod.keycloak.tool.keycloak_admin_health.anchor" {
 		t.Errorf("asked on %s", ToolSubject("keycloak", "keycloak_admin_health", "anchor"))
 	}
+	// A tool may answer that it waits for the operator (novox/hq ADR 0283): read with what it waits for, and not
+	// healthy, so nothing reading `healthy` alone takes it for a pass.
+	got, err := ReadToolAnswer([]byte(`{"result":{"state":"waits","healthy":false,"why":"the source games waits",` +
+		`"waits":[{"part":"the source games","secret":"smb-password-games","what":"the password of the source games"}]}}`))
+	if err != nil || got.Healthy || got.State != "waits" || len(got.Waits) != 1 || got.Waits[0].Secret != "smb-password-games" {
+		t.Fatalf("read as %+v (%v)", got, err)
+	}
+	h := ResourceHealth{Module: "mounts", Resource: "mounts.watch", State: StateWaiting, Waits: got.Waits}
+	b, _ := json.Marshal(h)
+	if !strings.Contains(string(b), `"state":"waiting"`) || !strings.Contains(string(b), `"waits":[{"part":"the source games","secret":"smb-password-games"`) {
+		t.Fatalf("said on the wire as %s", b)
+	}
 }
 
 // A declared tool check is asked on the link the node-engine actually holds — the NATS link — and not
@@ -199,18 +211,3 @@ func TestAToolCheckIsAskedOnTheLinkTheEngineHolds(t *testing.T) {
 		t.Fatalf("the NATS link was not asked: %v", err)
 	}
 }
-
-// A tool may answer that it waits for the operator (novox/hq ADR 0283): read with what it waits for, and not
-// healthy, so nothing reading `healthy` alone takes it for a pass.
-func TestAToolsWaitIsReadWithWhatItWaitsFor(t *testing.T) {
-	got, err := ReadToolAnswer([]byte(`{"result":{"state":"waits","healthy":false,"why":"the source games waits",` +
-		`"waits":[{"part":"the source games","secret":"smb-password-games","what":"the password of the source games"}]}}`))
-	if err != nil || got.Healthy || got.State != "waits" || len(got.Waits) != 1 || got.Waits[0].Secret != "smb-password-games" {
-		t.Fatalf("read as %+v (%v)", got, err)
-	}
-	h := ResourceHealth{Module: "mounts", Resource: "mounts.watch", State: StateWaiting, Waits: got.Waits}
-	b, _ := json.Marshal(h)
-	if !strings.Contains(string(b), `"state":"waiting"`) || !strings.Contains(string(b), `"waits":[{"part":"the source games","secret":"smb-password-games"`) {
-		t.Fatalf("said on the wire as %s", b)
-	}
-}

internal/selfcompose/data.go

@@ -0,0 +1,197 @@
+package selfcompose
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"sort"
+	"strings"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/compose"
+	"git.novox.be/novox/mesh-sdk/go/compose/frombus"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// **The data items of each module placed here, as this engine composed them** (novox/hq ADR 0577 clause 1). After a
+// pass that composed and handed something new (or the first one of a run), the composer works out each placed module's
+// data items from the same inputs it composed from (frombus.DataItems, which reads them through this composer's own
+// view of the bus) and keeps each module's in the engine's state data, key `<node>.<module>`, with the digest of the
+// module's composition: written only when the value differs, deleted when the module no longer declares any here or left
+// the node. That node's disk-load watches its node's keys and measures what they name; it composes nothing (clause 2).
+
+// DataStore is this node's keys of the engine's state data (link.EngineData).
+type DataStore interface {
+	// Get is a key's value, nil when it has none.
+	Get(ctx context.Context, key string) ([]byte, error)
+	Put(ctx context.Context, key string, value []byte) error
+	Delete(ctx context.Context, key string) error
+	// Keys is every key matching pattern that holds a value.
+	Keys(ctx context.Context, pattern string) ([]string, error)
+}
+
+// itemsReader is the composer's view of the bus for working out data items: no own secret is asked of the vault, since
+// working out where an item lies reads no secret (the items are rendered as a plan is, compose.Planning).
+type itemsReader struct{ reader }
+
+func (itemsReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow, error) {
+	return nil, nil
+}
+
+// dataItemsOf works out the data items of the scope's node (frombus.DataItems); a test sets what it answers.
+var dataItemsOf = frombus.DataItems
+
+// publishData keeps the data items of this node's modules in the engine's state data, when a pass asked it (dataDue).
+func (c *Composer) publishData(ctx context.Context) {
+	c.mu.Lock()
+	due := c.dataDue
+	c.dataDue = false
+	c.mu.Unlock()
+	if !due || c.env.Data == nil {
+		return
+	}
+	items, unplaced, err := dataItemsOf(ctx, itemsReader{reader{c: c}}, c.env.Mesh, frombus.Scope{Node: c.env.Node})
+	if err != nil {
+		c.sayData("the data items of this node's modules cannot be worked out, and what the engine's state " +
+			delivery.EngineDataBucket + " holds stands: " + err.Error())
+		return
+	}
+	if len(unplaced) > 0 {
+		c.env.Say("data items left out as not placeable: " + strings.Join(unplaced, "; "))
+	}
+	c.mu.Lock()
+	digests := c.digestsByModule()
+	c.mu.Unlock()
+	wrote, removed, err := c.writeData(ctx, items, digests)
+	if err != nil {
+		c.mu.Lock()
+		c.dataDue = true // the next pass tries again, whatever it hands
+		c.mu.Unlock()
+		c.sayData("the data items of this node's modules are not all kept in " + delivery.EngineDataBucket +
+			", and are written again at the next composition: " + err.Error())
+		return
+	}
+	c.sayData("")
+	if wrote+removed > 0 {
+		c.env.Say(fmt.Sprintf("the data items of this node's modules: %d key(s) of %s written, %d deleted",
+			wrote, delivery.EngineDataBucket, removed))
+	}
+}
+
+// sayData says a line about the data items once until it changes; "" clears it.
+func (c *Composer) sayData(line string) {
+	c.mu.Lock()
+	same := c.dataSaid == line
+	c.dataSaid = line
+	c.mu.Unlock()
+	if !same && line != "" {
+		c.env.Say(line)
+	}
+}
+
+// digestsByModule is the digest of each module's composition handed last, and the node's own under "" for a module run
+// inside it. Under c.mu.
+func (c *Composer) digestsByModule() map[string]string {
+	out := map[string]string{}
+	for key, dig := range c.handed {
+		_, module, _, isNode, err := delivery.ParseComposedKey(key)
+		if err != nil {
+			continue
+		}
+		if isNode {
+			module = ""
+		}
+		out[module] = dig
+	}
+	return out
+}
+
+// writeData writes each module's key whose value differs from what the key holds, and deletes this node's keys of
+// modules with no items here now. The keys are read once per run (dataKept), so a restart that finds the same items
+// writes nothing.
+func (c *Composer) writeData(ctx context.Context, items map[string][]catalogue.PlacedDataItem, digests map[string]string) (wrote, removed int, err error) {
+	store := c.env.Data
+	if c.dataKept == nil {
+		pattern, err := delivery.EngineDataNodeKeys(c.env.Node)
+		if err != nil {
+			return 0, 0, err
+		}
+		keys, err := store.Keys(ctx, pattern)
+		if err != nil {
+			return 0, 0, fmt.Errorf("this node's keys cannot be listed: %w", err)
+		}
+		kept := map[string][]byte{}
+		for _, k := range keys {
+			raw, err := store.Get(ctx, k)
+			if err != nil {
+				return 0, 0, fmt.Errorf("%s cannot be read: %w", k, err)
+			}
+			if raw != nil {
+				kept[k] = raw
+			}
+		}
+		c.dataKept = kept
+	}
+	want := map[string][]byte{}
+	modules := make([]string, 0, len(items))
+	for m := range items {
+		modules = append(modules, m)
+	}
+	sort.Strings(modules)
+	var failed []string
+	for _, module := range modules {
+		if len(items[module]) == 0 {
+			continue
+		}
+		key, err := delivery.EngineDataKey(c.env.Node, module)
+		if err != nil {
+			failed = append(failed, err.Error())
+			continue
+		}
+		digest, ok := digests[module]
+		if !ok {
+			digest = digests[""] // run inside the node's own composition
+		}
+		raw, err := delivery.EncodeEngineData(delivery.EngineData{Node: c.env.Node, Module: module, Digest: digest, Items: items[module]})
+		if err != nil {
+			failed = append(failed, err.Error())
+			continue
+		}
+		want[key] = raw
+	}
+	keys := make([]string, 0, len(want))
+	for k := range want {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	for _, key := range keys {
+		if bytes.Equal(c.dataKept[key], want[key]) {
+			continue
+		}
+		if err := store.Put(ctx, key, want[key]); err != nil {
+			failed = append(failed, key+": "+err.Error())
+			continue
+		}
+		c.dataKept[key] = want[key]
+		wrote++
+	}
+	var gone []string
+	for key := range c.dataKept {
+		if _, still := want[key]; !still {
+			gone = append(gone, key)
+		}
+	}
+	sort.Strings(gone)
+	for _, key := range gone {
+		if err := store.Delete(ctx, key); err != nil {
+			failed = append(failed, key+": "+err.Error())
+			continue
+		}
+		delete(c.dataKept, key)
+		removed++
+	}
+	if len(failed) > 0 {
+		return wrote, removed, fmt.Errorf("%s", strings.Join(failed, "; "))
+	}
+	return wrote, removed, nil
+}

internal/selfcompose/data_test.go

@@ -0,0 +1,148 @@
+package selfcompose
+
+import (
+	"context"
+	"errors"
+	"slices"
+	"strings"
+	"sync"
+	"testing"
+
+	"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"
+)
+
+// fakeData is the engine's state data as a map, counting the writes and deletes made.
+type fakeData struct {
+	mu      sync.Mutex
+	kv      map[string][]byte
+	puts    []string
+	deletes []string
+	fail    error
+}
+
+func (f *fakeData) Get(_ context.Context, key string) ([]byte, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.kv[key], nil
+}
+
+func (f *fakeData) Put(_ context.Context, key string, value []byte) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.fail != nil {
+		return f.fail
+	}
+	f.kv[key] = value
+	f.puts = append(f.puts, key)
+	return nil
+}
+
+func (f *fakeData) Delete(_ context.Context, key string) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	delete(f.kv, key)
+	f.deletes = append(f.deletes, key)
+	return nil
+}
+
+func (f *fakeData) Keys(_ context.Context, pattern string) ([]string, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	var out []string
+	for k := range f.kv {
+		if matches(pattern, k) {
+			out = append(out, k)
+		}
+	}
+	return out, nil
+}
+
+// ADR 0577 clause 1 (and its check in clause 5): after a pass that composed something new, the engine keeps each placed
+// module's data items in its state data, key `<node>.<module>`, with its composition's digest: written when the module's
+// items change, not written when they do not, and deleted when the module leaves the node; a key of another node is never
+// touched, a pass that handed nothing new reads nothing, and a write that failed is tried again at the next pass.
+func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) {
+	item := func(module, id, path string) catalogue.PlacedDataItem {
+		return catalogue.PlacedDataItem{Module: module, ID: id, Class: "valuable", Path: path, Measure: catalogue.MeasureWalk}
+	}
+	store := &fakeData{kv: map[string][]byte{
+		// Left by a module that went while the engine was down, and another node's key.
+		"n1.gone": []byte(`{"v":1}`), "n2.web": []byte(`{"v":1}`)}}
+	var items map[string][]catalogue.PlacedDataItem
+	var asked int
+	setFor(t, &dataItemsOf, func(context.Context, frombus.Reader, frombus.Mesh, frombus.Scope) (map[string][]catalogue.PlacedDataItem, []string, error) {
+		asked++
+		return items, nil, nil
+	})
+	// The engine's grant lets it read its own node's keys, as the listing at a run's first pass does, and no other's.
+	if !grantAllows(t, "n1", delivery.EngineDataBucket, "n1.*") || grantAllows(t, "n1", delivery.EngineDataBucket, "n2.*") {
+		t.Error("the node's grant reads other keys of the engine's state data than its own")
+	}
+	c := New(Env{Node: "n1", Data: store})
+	webKey, _ := delivery.ComposedKey("n1", "web", delivery.SourcePrefix2+strings.Repeat("a", 64))
+	nodeKey, _ := delivery.ComposedNodeKey("n1")
+	for _, step := range []struct {
+		name    string
+		due     bool
+		items   map[string][]catalogue.PlacedDataItem
+		digests map[string]string // composed key to digest handed
+		fail    error
+		puts    []string
+		deletes []string
+		asked   int
+		// want is each key's items and digest after the step.
+		want map[string]delivery.EngineData
+	}{
+		{name: "the run's first pass writes each module's items and deletes what left", due: true,
+			items:   map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db")}, "oldie": {item("oldie", "x", "/srv/oldie")}, "bare": nil},
+			digests: map[string]string{webKey: "sha256:w1", nodeKey: "sha256:n1"},
+			puts:    []string{"n1.oldie", "n1.web"}, deletes: []string{"n1.gone"}, asked: 1,
+			want: map[string]delivery.EngineData{
+				"n1.web":   {Module: "web", Digest: "sha256:w1", Items: []catalogue.PlacedDataItem{item("web", "db", "/srv/web/db")}},
+				"n1.oldie": {Module: "oldie", Digest: "sha256:n1", Items: []catalogue.PlacedDataItem{item("oldie", "x", "/srv/oldie")}}}},
+		{name: "a pass that handed nothing new reads nothing", due: false, asked: 1},
+		{name: "a pass that changes no module's items writes nothing", due: true,
+			items:   map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db")}, "oldie": {item("oldie", "x", "/srv/oldie")}},
+			digests: map[string]string{webKey: "sha256:w1", nodeKey: "sha256:n1"}, asked: 2},
+		{name: "a write that fails is tried again at the next pass", due: true, fail: errors.New("the bus said no"),
+			items:   map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db2")}, "oldie": {item("oldie", "x", "/srv/oldie")}},
+			digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, asked: 3},
+		{name: "a module whose composition changed its items is written again", due: false,
+			items:   map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db2")}, "oldie": {item("oldie", "x", "/srv/oldie")}},
+			digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, puts: []string{"n1.web"}, asked: 4,
+			want: map[string]delivery.EngineData{
+				"n1.web": {Module: "web", Digest: "sha256:w2", Items: []catalogue.PlacedDataItem{item("web", "db", "/srv/web/db2")}}}},
+		{name: "a module that left the node is deleted", due: true,
+			items:   map[string][]catalogue.PlacedDataItem{"oldie": {item("oldie", "x", "/srv/oldie")}},
+			digests: map[string]string{nodeKey: "sha256:n1"}, deletes: []string{"n1.web"}, asked: 5},
+	} {
+		t.Run(step.name, func(t *testing.T) {
+			store.mu.Lock()
+			store.puts, store.deletes, store.fail = nil, nil, step.fail
+			store.mu.Unlock()
+			items = step.items
+			c.mu.Lock()
+			c.handed = step.digests
+			c.dataDue = c.dataDue || step.due
+			c.mu.Unlock()
+			c.publishData(context.Background())
+			slices.Sort(store.puts)
+			if !slices.Equal(store.puts, step.puts) || !slices.Equal(store.deletes, step.deletes) || asked != step.asked {
+				t.Fatalf("wrote %v and deleted %v, items asked %d times; want %v, %v, %d", store.puts, store.deletes, asked,
+					step.puts, step.deletes, step.asked)
+			}
+			for key, want := range step.want {
+				got, err := delivery.DecodeEngineData(store.kv[key])
+				if err != nil || got.Node != "n1" || got.Module != want.Module || got.Digest != want.Digest ||
+					!slices.Equal(got.Items, want.Items) {
+					t.Errorf("%s holds %+v (%v), want %+v", key, got, err, want)
+				}
+			}
+			if store.kv["n2.web"] == nil {
+				t.Error("another node's key was deleted")
+			}
+		})
+	}
+}

internal/selfcompose/selfcompose.go

@@ -87,6 +87,9 @@ type Env struct {
 	// 0550 §2): asked only when a module placed here asks for a certificate and the record holds none from an authority
 	// the node trusts. Nil asks nothing, and such a module stands in, as before.
 	Certify func(ctx context.Context) error
+	// Data is this node's keys of the engine's state data (novox/hq ADR 0577 clause 1), where each placed module's data
+	// items are kept after a pass that composed something new. Nil keeps none.
+	Data DataStore
 }
 
 // VaultSecretChanged is the vault's event that a secret changed: a name and its generation, never a value.
@@ -146,6 +149,12 @@ type Composer struct {
 	last Pass
 	// watching says the composer's loop runs (Watch), so an ask of a pass can be answered.
 	watching atomic.Bool
+	// dataDue says a pass composed something new (or is the run's first), so the data items are kept again
+	// (publishData); dataSaid is the last line said of them. Under mu. dataKept is what this node's keys of the state
+	// data hold, read once per run, nil before; the loop's alone.
+	dataDue  bool
+	dataSaid string
+	dataKept map[string][]byte
 }
 
 // Pass is how one pass went, as the node-engine's verb compose answers it.
@@ -432,6 +441,7 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er
 					return
 				}
 			}
+			c.publishData(ctx)
 			if c.owed.Load() {
 				wait, _ := c.waitNow()
 				c.env.Say(fmt.Sprintf("the node composes again in %s: a read on the bus did not answer", wait))
@@ -894,6 +904,8 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
 		}
 	}
 	pass.Handed = len(out)
+	// The data items are kept again after a pass that handed something new, and after the run's first (ADR 0577 clause 1).
+	c.dataDue = c.dataDue || len(out) > 0 || c.dataKept == nil
 	return out
 }