mesh-provisioner composes from one watched copy of its reads, not key by key

Composing one node read every placement, have, module and resolved key one at a time and took past 30 s with the bus
local (ADR 0580 clause 2): the holder now reads them from mesh-sdk's watched copy, its watch handlers queue the nodes a
change touches after the copy took it, and a resolved value's change recomposes its node. mesh-sdk pinned at ba8b8f2.

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

modules/mesh-provisioner/cmd/mesh-provisioner/holder.go

@@ -29,6 +29,7 @@ import (
 	"sync"
 	"time"
 
+	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/asks"
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
 	"git.novox.be/novox/mesh-sdk/go/compose/frombus"
@@ -1044,12 +1045,52 @@ func orSaid(words ...string) string {
 	return ""
 }
 
-// FromBus is the holder's composer over the bus: frombus's plans of a node, at the builds keptAt names, read through
-// this module's runtime with its own `reads` (frombus.StatesReads). A plan names no value, so no own secret and no
-// credential is read (compose.Planning).
-func FromBus() Composer {
-	r := fromruntime.Reader()
+// FromWatched is the holder's composer over one watched copy of its reads (novox/hq ADR 0580 clause 2): frombus's plans
+// of a node, at the builds keptAt names, read from the copy (placement, haves, modules, resolved values), a build by key
+// and a manifest once per digest; no state is read key by key per composition. A plan names no value, so no own secret
+// and no credential is read (compose.Planning).
+func FromWatched(ws *stdio.WatchedStates) Composer {
+	r := fromruntime.WatchedReader(ws)
 	return func(ctx context.Context, node string, keptAt map[string]string) (map[string]catalogue.Plan, map[string]string, error) {
 		return frombus.PlansHeld(ctx, r, frombus.Mesh{}, frombus.Scope{Node: node, KeptAt: keptAt})
 	}
 }
+
+// WatchedReads are the states the holder reads whole, each from the watched copy: what the composer reads and the
+// wants (ADR 0580 clause 2). Its own states and the providers' entries stay read from the bus, after its own writes.
+var WatchedReads = []string{PlacementRead, WantsRead, ModulesRead, HaveRead, ResolvedRead}
+
+// Watching is a Store whose WatchedReads come from the copy; every other state, and every write, from s.
+func Watching(s Store, ws *stdio.WatchedStates) Store { return watchingStore{Store: s, ws: ws} }
+
+type watchingStore struct {
+	Store
+	ws *stdio.WatchedStates
+}
+
+func (w watchingStore) Get(state, key string) (*Entry, error) {
+	if !slices.Contains(WatchedReads, state) {
+		return w.Store.Get(state, key)
+	}
+	v, found, err := w.ws.Get(state, key)
+	if err != nil || !found {
+		return nil, err
+	}
+	return &Entry{Key: key, Value: v}, nil
+}
+
+func (w watchingStore) Keys(state string) ([]string, error) {
+	if !slices.Contains(WatchedReads, state) {
+		return w.Store.Keys(state)
+	}
+	return w.ws.Keys(state)
+}
+
+// NodesOfResolved is the node a resolved value's key `<node>.<module>.<key>` names: its change recomposes that node
+// (ADR 0580 clause 2).
+func NodesOfResolved(key string) []string {
+	if n, _, ok := strings.Cut(key, "."); ok && n != "" {
+		return []string{n}
+	}
+	return nil
+}

modules/mesh-provisioner/cmd/mesh-provisioner/holder_integration_test.go

@@ -3,8 +3,9 @@
 package main
 
 // The holder against a real bus (novox/hq ADR 0570 clauses 1-3, 5; issue 1954 H5): its states and every input it reads
-// are a real nats-server's key-value buckets, its plans composed by mesh-sdk's composition from bus state over the
-// in-memory mesh frombustest gives (restic on n1, requiring postgres-database from postgres there). A placement and a
+// are a real nats-server's key-value buckets, its plans composed by mesh-sdk's composition from the watched copy of those
+// buckets (ADR 0580 clause 2) filled with the mesh frombustest gives (restic on n1, requiring postgres-database from
+// postgres there); no input is read key by key per composition. A placement and a
 // want compose the key; the providers' outcomes (the vault's entry for the credential, postgres's own have) turn it
 // provided; restic leaving the node makes it retiring and raises the ask.
 
@@ -19,10 +20,12 @@ import (
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 
+	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/asks"
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
 	"git.novox.be/novox/mesh-sdk/go/compose/frombus"
 	"git.novox.be/novox/mesh-sdk/go/compose/frombus/frombustest"
+	"git.novox.be/novox/mesh-sdk/go/compose/fromruntime"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 	"git.novox.be/novox/mesh-sdk/go/provisions"
 	"git.novox.be/novox/mesh-sdk/go/testbus"
@@ -166,10 +169,101 @@ func TestTheHolderComposesProvidesAndAsksOnTheBus(t *testing.T) {
 	if err := s.Put(WantsRead, "want."+node+".restic", map[string]any{"v": 1, "fp": fp, "self": true}); err != nil {
 		t.Fatal(err)
 	}
+	// The rest of the mesh on the bus as its writers keep it: n1's placement, its record in its have, the builds.
+	putPlacement := func() {
+		raw, err := delivery.EncodePlacement(mesh.Placement[node])
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := s.Put(PlacementRead, "placement."+node, json.RawMessage(raw)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	putPlacement()
+	for _, n := range []string{node, "n2"} { // n2, where the vault runs, has its record too
+		rec := mesh.Record[node]
+		rec.Node = n
+		recRaw, err := delivery.EncodeHaveNode(delivery.HaveNode{Record: &rec})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := s.Put(HaveRead, mustKey(delivery.HaveNodeKey(n)), json.RawMessage(recRaw)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	for at, b := range mesh.Builds {
+		module, fp, _ := strings.Cut(at, " ")
+		raw, err := delivery.EncodeBuilderBuild(*b)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := s.Put(BuildsRead, mustKey(delivery.BuilderBuildKey(module, fp)), json.RawMessage(raw)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	// The watched copy over the real buckets: each state by one watch, a build and a manifest by key.
+	ws := stdio.WatchedStatesOver(func(state string, handle func(stdio.StateChange) error) error {
+		kv, err := s.kv(state)
+		if err != nil {
+			return err
+		}
+		w, err := kv.WatchAll(ctx)
+		if err != nil {
+			return err
+		}
+		hand := func(e jetstream.KeyValueEntry) {
+			op := stdio.OpPut
+			if e.Operation() != jetstream.KeyValuePut {
+				op = stdio.OpDelete
+			}
+			_ = handle(stdio.StateChange{State: state, Key: e.Key(), Op: op, Value: e.Value(), Revision: e.Revision()})
+		}
+		for e := range w.Updates() {
+			if e == nil { // the current values are all handed over
+				break
+			}
+			hand(e)
+		}
+		go func() {
+			for e := range w.Updates() {
+				hand(e)
+			}
+		}()
+		return nil
+	}, func(state, key string) ([]byte, bool, error) {
+		e, err := s.Get(state, key)
+		if err != nil || e == nil {
+			return nil, false, err
+		}
+		return e.Value, true, nil
+	})
+	// settled waits until the copy holds a key as the bus does: in the mesh the holder is woken by that same watch.
+	settled := func(state, key string) {
+		t.Helper()
+		for deadline := time.Now().Add(5 * time.Second); ; time.Sleep(10 * time.Millisecond) {
+			e, _ := s.Get(state, key)
+			v, found, _ := ws.Get(state, key)
+			if (e == nil && !found) || (e != nil && found && string(e.Value) == string(v)) {
+				return
+			}
+			if time.Now().After(deadline) {
+				t.Fatalf("the copy of %s never took %s", state, key)
+			}
+		}
+	}
+	fetch := func(_ context.Context, url string) ([]byte, error) {
+		for digest, body := range mesh.Archives {
+			if strings.HasSuffix(url, digest) {
+				return body, nil
+			}
+		}
+		return nil, errors.New("no archive at " + url)
+	}
+	reader := frombus.StatesReader(fromruntime.Watched{WatchedStates: ws}, fetch)
 	a := &asked{}
-	h := &Holder{Store: s, Asks: a, Now: time.Now, Logf: t.Logf,
+	h := &Holder{Store: Watching(s, ws), Asks: a, Now: time.Now, Logf: t.Logf,
 		Compose: func(ctx context.Context, node string, keptAt map[string]string) (map[string]catalogue.Plan, map[string]string, error) {
-			return frombus.PlansHeld(ctx, mesh, frombus.Mesh{}, frombus.Scope{Node: node, KeptAt: keptAt})
+			return frombus.PlansHeld(ctx, reader, frombus.Mesh{}, frombus.Scope{Node: node, KeptAt: keptAt})
 		}}
 	state := func(key string) provisions.Provision {
 		e, _ := s.Get(ProvisionsState, key)
@@ -199,6 +293,7 @@ func TestTheHolderComposesProvidesAndAsksOnTheBus(t *testing.T) {
 		"health": "ok"}); err != nil {
 		t.Fatal(err)
 	}
+	settled(HaveRead, node+".postgres")
 	for i := 0; i < 2; i++ { // the credential judged first, then what depends on it
 		if _, err := h.ComposeNode(ctx, node); err != nil {
 			t.Fatal(err)
@@ -217,9 +312,12 @@ func TestTheHolderComposesProvidesAndAsksOnTheBus(t *testing.T) {
 	}
 	p.Modules = kept
 	mesh.Placement[node] = p
+	putPlacement()
 	if err := s.Delete(WantsRead, "want."+node+".restic"); err != nil {
 		t.Fatal(err)
 	}
+	settled(PlacementRead, "placement."+node)
+	settled(WantsRead, "want."+node+".restic")
 	if _, err := h.ComposeNode(ctx, node); err != nil {
 		t.Fatal(err)
 	}

modules/mesh-provisioner/cmd/mesh-provisioner/holder_test.go

@@ -12,6 +12,7 @@ import (
 	"testing"
 	"time"
 
+	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/asks"
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -391,15 +392,55 @@ func TestAModuleHeldWithoutAWantIsComposedAtTheBuildItRuns(t *testing.T) {
 		}
 		return nil, nil, nil
 	}
-	c := m.compose(t, "novox")
-	if keptAt["nats"] != fp || len(keptAt) != 1 {
-		t.Fatalf("composed at %v, want nats at the build it runs and nothing of another node", keptAt)
+	// ADR 0580 clause 2: the reads come from one watched copy, each state watched once and none read key by key from the
+	// bus, and a change the copy takes wakes the node it touches (a resolved value its node).
+	bus := &readsOf{Store: m.s, read: map[string]int{}}
+	handles, watches := map[string]func(stdio.StateChange) error{}, map[string]int{}
+	ws := stdio.WatchedStatesOver(func(state string, handle func(stdio.StateChange) error) error {
+		watches[state]++
+		handles[state] = handle
+		keys, _ := m.s.Keys(state)
+		for _, k := range keys {
+			e, _ := m.s.Get(state, k)
+			_ = handle(stdio.StateChange{Key: k, Op: stdio.OpPut, Value: e.Value, Current: true})
+		}
+		return nil
+	}, func(string, string) ([]byte, bool, error) { return nil, false, nil })
+	m.h.Store = Watching(bus, ws)
+	wk := &work{pending: map[string]bool{}, wake: make(chan struct{}, 1)}
+	ws.Follow(ResolvedRead, func(c stdio.StateChange) error { wk.add(NodesOfResolved(c.Key)...); return nil })
+	for i := range 2 {
+		c := m.compose(t, "novox")
+		if keptAt["nats"] != fp || len(keptAt) != 1 {
+			t.Fatalf("composed at %v, want nats at the build it runs and nothing of another node", keptAt)
+		}
+		if i == 0 && !slices.Contains(c.Plans, "novox.nats") {
+			t.Fatalf("no plan written for a module held without a want: %+v", c)
+		}
 	}
-	if !slices.Contains(c.Plans, "novox.nats") {
-		t.Fatalf("no plan written for a module held without a want: %+v", c)
+	for _, r := range WatchedReads {
+		if bus.read[r] != 0 || watches[r] > 1 {
+			t.Errorf("%s was read %d times from the bus and watched %d times", r, bus.read[r], watches[r])
+		}
+	}
+	_ = handles[ResolvedRead](stdio.StateChange{Key: "g14.dunst.font-size", Op: stdio.OpPut, Value: []byte("13")})
+	if !wk.pending["g14"] || len(wk.pending) != 1 {
+		t.Errorf("a resolved value's change queued %v, want g14", wk.pending)
 	}
 }
 
+// readsOf counts the reads that reach the bus, by state.
+type readsOf struct {
+	Store
+	read map[string]int
+}
+
+func (r *readsOf) Get(state, key string) (*Entry, error) {
+	r.read[state]++
+	return r.Store.Get(state, key)
+}
+func (r *readsOf) Keys(state string) ([]string, error) { r.read[state]++; return r.Store.Keys(state) }
+
 // The holder holds no ticker: it composes on change events alone (ADR 0570 clause 2; issue 1954 H1).
 func TestTheHolderHoldsNoTimer(t *testing.T) {
 	files, _ := filepath.Glob("*.go")

modules/mesh-provisioner/cmd/mesh-provisioner/main.go

@@ -105,13 +105,16 @@ func (w *work) run(h *Holder) {
 
 func main() {
 	s := runtimeStore{}
-	h := &Holder{Store: s, Compose: FromBus(), Asks: runtimeAsks{}, Now: time.Now, Logf: logf}
+	ws := stdio.NewWatchedStates() // one copy of what it reads, each state by its watch (ADR 0580 clause 2)
+	h := &Holder{Store: Watching(s, ws), Compose: FromWatched(ws), Asks: runtimeAsks{}, Now: time.Now, Logf: logf}
 	w := &work{pending: map[string]bool{}, wake: make(chan struct{}, 1)}
 	go w.run(h)
 	go starting(h)
-	go stdio.State(PlacementRead).Follow("", func(c stdio.StateChange) error { w.add(NodesOfPlacement(c.Key)...); return nil })
-	go stdio.State(WantsRead).Follow("", func(c stdio.StateChange) error { w.add(NodesOfWant(c.Key)...); return nil })
-	go stdio.State(ModulesRead).Follow("", func(c stdio.StateChange) error {
+	// Each watch's handler runs after the copy took its change, so a composition it wakes reads the copy that change made.
+	go ws.Follow(PlacementRead, func(c stdio.StateChange) error { w.add(NodesOfPlacement(c.Key)...); return nil })
+	go ws.Follow(WantsRead, func(c stdio.StateChange) error { w.add(NodesOfWant(c.Key)...); return nil })
+	go ws.Follow(ResolvedRead, func(c stdio.StateChange) error { w.add(NodesOfResolved(c.Key)...); return nil })
+	go ws.Follow(ModulesRead, func(c stdio.StateChange) error {
 		if module, err := delivery.ParseModulesKey(c.Key); err == nil {
 			w.add(h.NodesOfModule(module)...)
 		}