From aa4a4253ad2415cbaaf0027c9cdbec5a4918bf4f Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 1 Nov 2026 03:31:17 +0100 Subject: [PATCH] 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 --- .../cmd/mesh-provisioner/holder.go | 51 ++++++++- .../holder_integration_test.go | 106 +++++++++++++++++- .../cmd/mesh-provisioner/holder_test.go | 51 ++++++++- .../cmd/mesh-provisioner/main.go | 11 +- 4 files changed, 201 insertions(+), 18 deletions(-) diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/holder.go b/modules/mesh-provisioner/cmd/mesh-provisioner/holder.go index 1242c101..14769fc4 100644 --- a/modules/mesh-provisioner/cmd/mesh-provisioner/holder.go +++ b/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 `..` 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 +} diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/holder_integration_test.go b/modules/mesh-provisioner/cmd/mesh-provisioner/holder_integration_test.go index 4b650470..708f4571 100644 --- a/modules/mesh-provisioner/cmd/mesh-provisioner/holder_integration_test.go +++ b/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) } diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/holder_test.go b/modules/mesh-provisioner/cmd/mesh-provisioner/holder_test.go index 35cb4626..17e770a1 100644 --- a/modules/mesh-provisioner/cmd/mesh-provisioner/holder_test.go +++ b/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") diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/main.go b/modules/mesh-provisioner/cmd/mesh-provisioner/main.go index fda63ee2..f7467950 100644 --- a/modules/mesh-provisioner/cmd/mesh-provisioner/main.go +++ b/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)...) }