From 692f18ef3bde53983be63710631207116c870c66 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 12:08:56 +0100 Subject: [PATCH] CAT-C11: mesh-config reads the indexer's and the assigner's state through their tools and views events Six states (modules, manifests, predefined, config-candidates, placement, wants) are asked of the owners' read tools at start and kept from their views. events, a gap asks again; a reader started while an owner is down keeps asking and writes a finding past two minutes. The builder's builds has no owner tool and stays a read. The returning node-engine's event is read by its node token (ADR 0618 as built). Tools config_first_views and config_first_view_again. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- modules/mesh-config/cmd/mesh-config/main.go | 60 +++-- .../cmd/mesh-config/manifest_test.go | 6 + .../mesh-config/cmd/mesh-config/ownerviews.go | 249 ++++++++++++++++++ .../cmd/mesh-config/ownerviews_test.go | 76 ++++++ .../cmd/mesh-config/whole_integration_test.go | 124 ++++++++- modules/mesh-config/go.mod | 1 + modules/mesh-config/module.json | 28 +- modules/mesh-vault/cmd/mesh-vault/main.go | 37 ++- .../cmd/mesh-vault/mint_integration_test.go | 74 ++++++ .../cmd/mesh-vault/placementview.go | 182 +++++++++++++ .../mesh-vault/rotation_integration_test.go | 1 + modules/mesh-vault/cmd/mesh-vault/tools.go | 2 +- .../mesh-vault/cmd/mesh-vault/vault_test.go | 4 +- modules/mesh-vault/go.mod | 1 + modules/mesh-vault/module.json | 13 +- 15 files changed, 798 insertions(+), 60 deletions(-) create mode 100644 modules/mesh-config/cmd/mesh-config/ownerviews.go create mode 100644 modules/mesh-config/cmd/mesh-config/ownerviews_test.go create mode 100644 modules/mesh-vault/cmd/mesh-vault/placementview.go diff --git a/modules/mesh-config/cmd/mesh-config/main.go b/modules/mesh-config/cmd/mesh-config/main.go index 899b30c3..454126b5 100644 --- a/modules/mesh-config/cmd/mesh-config/main.go +++ b/modules/mesh-config/cmd/mesh-config/main.go @@ -10,6 +10,7 @@ package main import ( + "context" "encoding/json" "errors" "fmt" @@ -30,10 +31,19 @@ const ( func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) } -// runtimeStore is the bus, through the runtime that launched this bundle. +// theViews are the pictures of the delivery owners' state, made when the module runs (nil in a test of the tools). +var theViews *ownerViews + +// runtimeStore is the bus, through the runtime that launched this bundle. A state of the delivery owners that the +// owners' tools answer is read from the picture those keep (ownerviews.go), the rest from the bus. type runtimeStore struct{} func (runtimeStore) Get(state, key string) (*Entry, error) { + if theViews != nil { + if e, followed, err := theViews.get(state, key); followed { + return e, err + } + } e, err := stdio.State(state).Get(key) if err != nil || e == nil { return nil, err @@ -48,7 +58,14 @@ func (runtimeStore) Put(state, key string, value any) error { func (runtimeStore) Delete(state, key string) error { return stdio.State(state).Delete(key) } -func (runtimeStore) Keys(state string) ([]string, error) { return stdio.State(state).Keys() } +func (runtimeStore) Keys(state string) ([]string, error) { + if theViews != nil { + if keys, followed, err := theViews.keys(state); followed { + return keys, err + } + } + return stdio.State(state).Keys() +} func (runtimeStore) Append(log, key string, value any) error { _, err := stdio.Log(log).Append(key, value) @@ -102,7 +119,19 @@ func (runtimeAsks) Warranted(ask string) error { func main() { c := &Config{Store: runtimeStore{}, Logf: logf, Asks: runtimeAsks{}, Forge: toolForge{ask: stdio.Ask}, Delivery: toolDelivery{ask: stdio.Ask}, Vault: toolVault{ask: stdio.Ask, recorder: recorder}} - go resolving(c) + moved := make(chan struct{}, 1) + theViews = newOwnerViews(stdio.Ask, func() { + select { + case moved <- struct{}{}: + default: + } + }, func(key string, v json.RawMessage) error { + _, err := stdio.State(FindingsState).Put(key, v) + return err + }, + func(key string) error { return stdio.State(FindingsState).Delete(key) }) + theViews.Start(context.Background()) + go resolving(c, moved) go hearing(c) go hearingDelivery(c) if err := stdio.Serve("", Tools(c)); err != nil { @@ -114,22 +143,9 @@ func main() { // resolving resolves at start and again whenever a source moves, and every five minutes besides, so a change // heard nowhere is still resolved. The runtime answers only once Serve runs, so the watches are asked again // until they are taken. -func resolving(c *Config) { - moved := make(chan struct{}, 1) - poke := func(stdio.StateChange) error { - select { - case moved <- struct{}{}: - default: - } - return nil - } - // A build's manifest is written before a want names the build, so a want, a trunk module or a placement moving is - // every change of what a module declares where (issue 1953); the builds themselves are read, not watched. - for _, read := range []string{PlacementRead, WantsRead, ModulesRead, IndexerPredefinedRead, - CandidatesRead} { - // Asked again until the runtime takes it, each give-up said (mesh-sdk's Follow). - go stdio.State(read).Follow("", poke) - } +func resolving(c *Config, moved chan struct{}) { + // The owners' states reach `moved` through their views (ownerviews.go); the builder's builds, which has no tool yet, + // is read when a want moves, and nothing else is watched. tick := time.NewTicker(5 * time.Minute) defer tick.Stop() time.Sleep(time.Second) @@ -239,6 +255,10 @@ func hearingDelivery(c *Config) { Node string `json:"node"` } _ = json.Unmarshal(e.Body, &b) + if b.Node == "" { + // The event's subject ends with the node's token (ADR 0618 as built): mesh.seat.node-engine.event.manifest-reads-changed.. + b.Node = e.Key[strings.LastIndex(e.Key, ".")+1:] + } return c.Windows(b.Node) }}, {"mesh-delivery-indexer.manifest-accepted", func(e stdio.Envelope) []string { @@ -302,7 +322,7 @@ func Tools(c *Config) []stdio.Tool { rows, err := c.Renames(strings.TrimSpace(m), true) return map[string]any{"renames": seatRenames(rows)}, err }} - return append(switchTools(c), renames, stdio.Tool{ + return append(append(switchTools(c), viewTools(c, func() *ownerViews { return theViews })...), renames, stdio.Tool{ Name: Seat + "." + Verb, Description: "Set, clear, list or read the history of config values (novox/hq ADR 0348). A value is set at " + "one level: mesh config (no module, no node) or node config (a node) for a predefined key, or a " + diff --git a/modules/mesh-config/cmd/mesh-config/manifest_test.go b/modules/mesh-config/cmd/mesh-config/manifest_test.go index b93fe457..42fa4e6d 100644 --- a/modules/mesh-config/cmd/mesh-config/manifest_test.go +++ b/modules/mesh-config/cmd/mesh-config/manifest_test.go @@ -40,6 +40,7 @@ func TestEveryStateAndActHasItsTool(t *testing.T) { // design 32: each new state with the tool that reads it, each act with the tool that starts it. RenamesState: RenamesTool, RetiringState: RetiringTool, FindingsState: "config_renames and config_retiring", "a move refused": RenameAgainTool, "a window dropped": RenameCloseTool, "an ask again": RetiringTool, + "the owners' first views": FirstViewsTool, "a first view asked again": FirstViewAgainTool, } served := map[string]bool{} for _, tool := range Tools(&Config{}) { @@ -73,6 +74,11 @@ func TestEveryStateAndActHasItsTool(t *testing.T) { t.Errorf("the manifest lists the tool %s, which is not served", name) } } + // The delivery owners' state is read through their tools and views (ADR 0615 §1): only the builder's builds, which has + // no tool yet, is still a read. + if len(m.Reads) != 1 || m.Reads[0] != BuildsRead { + t.Errorf("reads %v, want the builder's builds alone", m.Reads) + } for _, list := range [][]string{m.State, m.Logs, m.Reads} { for _, s := range list { if _, ok := via[s]; !ok { diff --git a/modules/mesh-config/cmd/mesh-config/ownerviews.go b/modules/mesh-config/cmd/mesh-config/ownerviews.go new file mode 100644 index 00000000..d7367e34 --- /dev/null +++ b/modules/mesh-config/cmd/mesh-config/ownerviews.go @@ -0,0 +1,249 @@ +package main + +// What mesh-config reads of the delivery owners' state, through their tools and views families (novox/hq ADR 0615 §1, +// §4 to §6, ADR 0596): never a bucket of theirs. For each of the six states it follows (the indexer's modules, manifests, +// predefined and config-candidates, the assigner's placement and wants) it asks the owner's read tool for the whole +// state at its start, keeps the picture from the owner's `views.` events, and asks again only on a gap. Started +// while an owner is down it keeps asking, and past two minutes writes a finding naming the owner and the tool. +// +// The ordinary read tools are asked, not the first-view tools: those answer a few named callers (mesh-grants, +// node-tools, the delivery parts), and mesh-config is not one. The builder's `builds` has no owner tool yet (A2 built the +// indexer's and the assigner's), so it is still read as a `reads` entry. + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" + + "git.novox.be/novox/mesh-delivery/sdk/views" +) + +// The tools of the owners that mesh-config asks, by the state they read. +const ( + IndexerModulesTool = "indexer_read_modules" + IndexerManifestsTool = "indexer_read_manifests" + IndexerPredefTool = "indexer_read_predefined" + IndexerCandidatesTool = "indexer_read_config_candidates" + AssignerPlacementTool = "assigner_read_placement" + AssignerWantsTool = "assigner_read_wants" + + FirstViewsTool = "config_first_views" + FirstViewAgainTool = "config_first_view_again" +) + +// followed is each state read through its owner: the `.` name this module's code uses, the owner, its +// state and its tool. +var followed = []struct{ Read, Owner, State, Tool string }{ + {ModulesRead, delivery.IndexerModule, "modules", IndexerModulesTool}, + {ManifestsRead, delivery.IndexerModule, "manifests", IndexerManifestsTool}, + {IndexerPredefinedRead, delivery.IndexerModule, delivery.IndexerPredefinedState, IndexerPredefTool}, + {CandidatesRead, delivery.IndexerModule, delivery.IndexerConfigCandidatesState, IndexerCandidatesTool}, + {PlacementRead, delivery.AssignerModule, "placement", AssignerPlacementTool}, + {WantsRead, delivery.AssignerModule, "wants", AssignerWantsTool}, +} + +// ownerViews are the pictures, by the read name. +type ownerViews struct { + by map[string]*views.Follower + // store keeps the findings a first view unanswered writes. + store func(key string, value json.RawMessage) error + drop func(key string) error +} + +// newOwnerViews makes a follower for each state; changed is told when any picture changes. +func newOwnerViews(ask func(key string, body any) (json.RawMessage, error), changed func(), put func(string, json.RawMessage) error, + del func(string) error) *ownerViews { + o := &ownerViews{by: map[string]*views.Follower{}, store: put, drop: del} + for _, f := range followed { + f := f + fol := &views.Follower{Owner: f.Owner, Tool: f.Owner + "." + f.Tool, State: f.State, Logf: logf, + Ask: func(_ context.Context, after string) (views.Answer, error) { + body := map[string]any{"values": true} + if after != "" { + body["after"] = after + } + raw, err := ask(f.Owner+"."+f.Tool, body) + if err != nil { + return views.Answer{}, err + } + answer, output, ok := unwrapAnswer(raw) + if !ok { + return views.Answer{}, fmt.Errorf("%s refused: %s", f.Tool, output) + } + return views.DecodeAnswer(answer) + }, + Changed: func(string, json.RawMessage, bool) { changed() }} + fol.Raise = func(s views.Status) { + if k, v, err := views.UnansweredFinding(Module, s, time.Now()); err == nil { + _ = put(k, v) + } + } + fol.Clear = func(s views.Status) { + if k, _, err := views.UnansweredFinding(Module, s, time.Now()); err == nil { + _ = del(k) + } + } + o.by[f.Read] = fol + } + return o +} + +// Start asks every first view until answered (each on its goroutine) and subscribes to the owners' views events. +func (o *ownerViews) Start(ctx context.Context) { + for _, fol := range o.by { + go func(f *views.Follower) { + if err := f.Start(ctx); err != nil { + logf("[mesh-config] the first view of %s's %s ended unanswered: %v", f.Owner, f.State, err) + } + }(fol) + } + for _, f := range followed { + f := f + pattern := f.Owner + "." + views.EventName(f.State) + go func() { + <-stdio.Served() + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + err := stdio.Subscribe(pattern, func(e stdio.Envelope) error { + ev, err := views.DecodeEvent(e.Body) + if err != nil { + logf("[mesh-config] %s is not a views event: %v", pattern, err) + return nil + } + fol := o.by[f.Read] + if gap, why := fol.Apply(ev); gap { + logf("[mesh-config] a gap in %s's %s (%s): its first view is asked once more", f.Owner, f.State, why) + go func() { _ = fol.Start(ctx) }() + } + return nil + }) + if err == nil { + return + } + logf("[mesh-config] does not hear %s yet (%v); asking again in %s", pattern, err, wait) + time.Sleep(wait) + } + }() + } +} + +// read is the picture of a read name, and whether the name is one followed. +func (o *ownerViews) read(state string) (*views.Follower, bool) { + f, ok := o.by[state] + return f, ok +} + +// errUnanswered says a first view is not answered yet: a reader waits, and its next resolution reads it. +func errUnanswered(f *views.Follower) error { + s := f.Status() + return fmt.Errorf("the first view of %s's %s is not answered yet (%s asked %d times)", s.Owner, s.State, s.Tool, s.Asks) +} + +// get is one key of a followed state. +func (o *ownerViews) get(state, key string) (*Entry, bool, error) { + f, ok := o.read(state) + if !ok { + return nil, false, nil + } + select { + case <-f.Ready(): + default: + return nil, true, errUnanswered(f) + } + v, held := f.Get(key) + if !held { + return nil, true, nil + } + return &Entry{Key: key, Value: v}, true, nil +} + +// keys are the keys of a followed state, sorted. +func (o *ownerViews) keys(state string) ([]string, bool, error) { + f, ok := o.read(state) + if !ok { + return nil, false, nil + } + select { + case <-f.Ready(): + default: + return nil, true, errUnanswered(f) + } + var out []string + for k := range f.Values() { + out = append(out, k) + } + sort.Strings(out) + return out, true, nil +} + +// statuses are the first views' statuses, by owner and state. +func (o *ownerViews) statuses() []views.Status { + var out []views.Status + for _, f := range followed { + out = append(out, o.by[f.Read].Status()) + } + return out +} + +// again asks one first view once more by hand (".", or all when empty). +func (o *ownerViews) again(ctx context.Context, which, why string) ([]string, error) { + var done []string + for _, f := range followed { + if which != "" && which != f.Owner+"."+f.State { + continue + } + if err := o.by[f.Read].Again(ctx, why); err != nil { + return done, err + } + done = append(done, f.Owner+"."+f.State) + } + if len(done) == 0 { + return nil, errors.New("no first view is named " + which) + } + return done, nil +} + +// viewTools are the read tool of the first views and its re-read (design 25 § 12.2): the status of each, and a hand ask. +func viewTools(c *Config, o func() *ownerViews) []stdio.Tool { + return []stdio.Tool{ + {Name: FirstViewsTool, + Description: "mesh-config's first views of the delivery owners' state: for each of the indexer's modules, manifests, " + + "predefined and config-candidates and the assigner's placement and wants, the tool it asks, the revision its " + + "picture holds, when it asked and was answered, how many asks (and by hand), and whether its condition is raised.", + Run: func(map[string]any) (any, error) { + ov := o() + if ov == nil { + return nil, errors.New("mesh-config reads its owners through views only when it runs as the module") + } + return map[string]any{"first-views": ov.statuses()}, nil + }}, + {Name: FirstViewAgainTool, + Description: "Ask a first view once more by hand, `.` (all when not given), recorded as a hand act with " + + "its why.", + Input: map[string]any{"type": "object", "required": []string{"why"}, "properties": map[string]any{ + "view": map[string]any{"type": "string", "description": "., such as mesh-delivery-assigner.wants"}, + "why": map[string]any{"type": "string"}}}, + Run: func(a map[string]any) (any, error) { + ov := o() + if ov == nil { + return nil, errors.New("mesh-config reads its owners through views only when it runs as the module") + } + why, _ := a["why"].(string) + view, _ := a["view"].(string) + if strings.TrimSpace(why) == "" { + return nil, errors.New("say why: a hand act keeps its why") + } + if err := c.Store.HandAct("config_first_view_again "+view, strings.TrimSpace(why), "config_first_view_again"); err != nil { + return nil, fmt.Errorf("the hand-act log does not hold it, so nothing is asked: %w", err) + } + done, err := ov.again(context.Background(), strings.TrimSpace(view), strings.TrimSpace(why)) + return map[string]any{"asked-again": done}, err + }}, + } +} diff --git a/modules/mesh-config/cmd/mesh-config/ownerviews_test.go b/modules/mesh-config/cmd/mesh-config/ownerviews_test.go new file mode 100644 index 00000000..8f04825f --- /dev/null +++ b/modules/mesh-config/cmd/mesh-config/ownerviews_test.go @@ -0,0 +1,76 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "strings" + "sync/atomic" + "testing" + "time" + + "git.novox.be/novox/mesh-delivery/sdk/views" +) + +// CAT-C11 (ADR 0615 §1): a followed state of the delivery owners is read from the picture of its owner's tool and events: +// a key asked before the first view is answered is an error (a reader waits), never a read of the owner's bucket; after +// it, the owner's value; an event at the next revision changes it, one past it is a gap and asks the first view again; a +// state that is no owner's falls through to the bus. +func TestAFollowedStateIsReadFromItsOwnersFirstViewAndEvents(t *testing.T) { + created := time.Date(2026, 11, 4, 9, 0, 0, 0, time.UTC) + var asks atomic.Int32 + hold := make(chan struct{}) + ask := func(key string, body any) (json.RawMessage, error) { + if key != "mesh-delivery-assigner.assigner_read_wants" || body.(map[string]any)["values"] != true { + return nil, errors.New("asked " + key) + } + <-hold + n := asks.Add(1) + a := views.Answer{V: 1, Owner: "mesh-delivery-assigner", State: "wants", Revision: uint64(n) * 10, Created: created, + Values: map[string]json.RawMessage{"want.g14.dunst": json.RawMessage(`{"fp":"a"}`)}} + raw, _ := json.Marshal(a) + return raw, nil + } + var changed atomic.Int32 + o := newOwnerViews(ask, func() { changed.Add(1) }, func(string, json.RawMessage) error { return nil }, func(string) error { return nil }) + if _, followed, err := o.get(WantsRead, "want.g14.dunst"); !followed || err == nil || !strings.Contains(err.Error(), "not answered yet") { + t.Fatalf("before the first view: followed %v, %v", followed, err) + } + if _, followed, _ := o.get(BuildsRead, "x"); followed { + t.Error("the builder's builds is followed, and it has no owner tool") + } + ctx := context.Background() + go func() { _ = o.by[WantsRead].Start(ctx) }() + hold <- struct{}{} + <-o.by[WantsRead].Ready() + e, isFollowed, err := o.get(WantsRead, "want.g14.dunst") + if err != nil || !isFollowed || e == nil || string(e.Value) != `{"fp":"a"}` { + t.Fatalf("after the first view: %+v, %v, %v", e, isFollowed, err) + } + ev := func(rev uint64, key string, deleted bool) views.Event { + return views.Event{V: 1, Owner: "mesh-delivery-assigner", State: "wants", Key: key, Revision: rev, Created: created, Deleted: deleted, + Value: json.RawMessage(`{"fp":"b"}`)} + } + if gap, why := o.by[WantsRead].Apply(ev(11, "want.novox.dunst", false)); gap { + t.Fatalf("the next revision is a gap: %s", why) + } + if keys, _, _ := o.keys(WantsRead); strings.Join(keys, ",") != "want.g14.dunst,want.novox.dunst" { + t.Errorf("keys %v", keys) + } + if gap, _ := o.by[WantsRead].Apply(ev(11, "want.novox.dunst", false)); gap { + t.Error("an event already in the picture is a gap") + } + if gap, why := o.by[WantsRead].Apply(ev(13, "want.x.y", false)); !gap || !strings.Contains(why, "changes missed") { + t.Errorf("a missed revision: gap %v, %s", gap, why) + } + o.by[WantsRead].Apply(ev(12, "want.g14.dunst", true)) + if e, _, _ := o.get(WantsRead, "want.g14.dunst"); e != nil { + t.Error("a deleted key is still held") + } + if changed.Load() < 3 { + t.Errorf("the resolution was told of %d changes", changed.Load()) + } + if got := o.statuses(); len(got) != len(followed) { + t.Errorf("statuses %d, want %d", len(got), len(followed)) + } +} diff --git a/modules/mesh-config/cmd/mesh-config/whole_integration_test.go b/modules/mesh-config/cmd/mesh-config/whole_integration_test.go index ebad48c3..4aeb182b 100644 --- a/modules/mesh-config/cmd/mesh-config/whole_integration_test.go +++ b/modules/mesh-config/cmd/mesh-config/whole_integration_test.go @@ -62,6 +62,93 @@ type player struct { readers []ReaderRow said []saidOnBus subs []string + // owned is each delivery owner's state as the owner's tool answers it: its values and the revision of its last change. + owned map[string]*ownedState +} + +// ownedState is one state of an owner (the indexer's modules, the assigner's wants...), kept from its bucket. +type ownedState struct { + owner, state string + revision uint64 + values map[string]json.RawMessage +} + +var theCreated = time.Date(2026, 11, 4, 9, 0, 0, 0, time.UTC) + +// own follows the bucket of an owner's state, and says each change after the first listing on the owner's views family +// to the module (once it subscribed), as the owner does (ADR 0615 §4). +func (p *player) own(read string) { + p.t.Helper() + owner, state, _ := strings.Cut(read, ".") + kv, err := p.kv(read) + if err != nil { + p.t.Fatal(err) + } + w, err := kv.WatchAll(p.ctx) + if err != nil { + p.t.Fatal(err) + } + st := &ownedState{owner: owner, state: state, values: map[string]json.RawMessage{}} + p.owned[read] = st + listed := make(chan struct{}) + go func() { + current := true + for e := range w.Updates() { + if e == nil { + current = false + close(listed) + continue + } + p.mu.Lock() + deleted := e.Operation() != jetstream.KeyValuePut + if deleted { + delete(st.values, e.Key()) + } else { + st.values[e.Key()] = append(json.RawMessage(nil), e.Value()...) + } + if !current { + st.revision++ + ev := map[string]any{"v": 1, "owner": owner, "state": state, "key": e.Key(), "revision": st.revision, "created": theCreated} + if deleted { + ev["deleted"] = true + } else { + ev["value"] = json.RawMessage(e.Value()) + } + p.next++ + id := fmt.Sprintf("o-%d", p.next) + body, _ := json.Marshal(ev) + raw, _ := json.Marshal(map[string]any{"id": id, "jsonrpc": "2.0", "method": "mesh/event", "params": map[string]any{ + "envelope": map[string]any{"key": owner + ".views." + state, "node": "novox", "body": json.RawMessage(body)}}}) + _, _ = p.in.Write(append(raw, '\n')) + } + p.mu.Unlock() + } + }() + select { + case <-listed: + case <-time.After(10 * time.Second): + p.t.Fatalf("%s was not listed", read) + } +} + +// ownerAnswer is an owner's read tool asked with values: the state whole at its revision. +func (p *player) ownerAnswer(key string) (any, bool) { + for read, st := range p.owned { + if key == st.owner+"."+toolOf(st) { + _ = read + return map[string]any{"v": 1, "owner": st.owner, "state": st.state, "revision": st.revision, "created": theCreated, + "values": st.values, "at": time.Now()}, true + } + } + return nil, false +} + +func toolOf(st *ownedState) string { + prefix := "indexer_read_" + if st.owner == "mesh-delivery-assigner" { + prefix = "assigner_read_" + } + return prefix + strings.ReplaceAll(st.state, "-", "_") } type saidOnBus struct { @@ -263,6 +350,9 @@ func (p *player) handle(m wire) { func (p *player) asked(key string, body json.RawMessage) any { p.mu.Lock() defer p.mu.Unlock() + if a, ok := p.ownerAnswer(key); ok { + return a + } switch { case key == HavesAsk: var q struct{ Node, Module string } @@ -407,6 +497,32 @@ func (p *player) get(state, key string) ([]byte, bool) { return e.Value(), true } +// viewsReady waits until every first view of the module is answered (config_first_views). +func (p *player) viewsReady() { + p.t.Helper() + eventually(p.t, "every first view is answered", func() bool { + raw, err := p.call(FirstViewsTool, nil) + if err != nil { + return false + } + var a struct { + Views []struct { + Revision uint64 `json:"revision"` + Answered time.Time `json:"Answered"` + } `json:"first-views"` + } + if json.Unmarshal(raw, &a) != nil || len(a.Views) != len(followed) { + return false + } + for _, v := range a.Views { + if v.Answered.IsZero() { + return false + } + } + return true + }) +} + func eventually(t *testing.T, what string, ok func() bool) { t.Helper() for deadline := time.Now().Add(30 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { @@ -475,11 +591,15 @@ func start(t *testing.T, bin string, js jetstream.JetStream, router *standIn, ha stdin, _ := cmd.StdinPipe() stdout, _ := cmd.StdoutPipe() cmd.Stderr = os.Stderr + p := &player{t: t, ctx: ctx, js: js, router: router, in: stdin, pending: map[string]chan wire{}, haves: haves, readers: readers, + owned: map[string]*ownedState{}} + for _, f := range followed { + p.own(f.Read) + } if err := cmd.Start(); err != nil { t.Fatal(err) } t.Cleanup(func() { _ = stdin.Close(); cancel(); _ = cmd.Wait() }) - p := &player{t: t, ctx: ctx, js: js, router: router, in: stdin, pending: map[string]chan wire{}, haves: haves, readers: readers} go p.serve(stdout) t.Cleanup(cancel) return p @@ -519,6 +639,7 @@ func TestConfigWholeRenamesAndRetiresAgainstTheBus(t *testing.T) { router := newStandIn() haves := map[string]string{"g14.player": fp1, "novox.player": fp1, "g14.radio": fp1} p := start(t, bin, js, router, haves, nil) + p.viewsReady() // The mesh: player and radio on g14, player also on novox, all on their first build; dock declares both names. const oldRadio = `{"module":"radio","config":{"band":{"type":"number","default":1}}}` @@ -583,6 +704,7 @@ func TestConfigWholeRenamesAndRetiresAgainstTheBus(t *testing.T) { cancelOld := p.ctx _ = cancelOld p2 := start(t, bin, js, router, haves, nil) + p2.viewsReady() if _, err := p2.call(ResyncTool, nil); err != nil { t.Fatal(err) } diff --git a/modules/mesh-config/go.mod b/modules/mesh-config/go.mod index aae59185..d781033f 100644 --- a/modules/mesh-config/go.mod +++ b/modules/mesh-config/go.mod @@ -3,6 +3,7 @@ module mesh-config go 1.26.0 require ( + git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102191058-f02f142af128 github.com/nats-io/nats.go v1.54.0 diff --git a/modules/mesh-config/module.json b/modules/mesh-config/module.json index ee35885e..f52841dc 100644 --- a/modules/mesh-config/module.json +++ b/modules/mesh-config/module.json @@ -27,7 +27,13 @@ "gitea.forge.pull-request.closed", "mesh-delivery-view.delivered", "node-engine.manifest-reads-changed", - "mesh-delivery-indexer.manifest-accepted" + "mesh-delivery-indexer.manifest-accepted", + "mesh-delivery-indexer.views.modules", + "mesh-delivery-indexer.views.manifests", + "mesh-delivery-indexer.views.predefined", + "mesh-delivery-indexer.views.config-candidates", + "mesh-delivery-assigner.views.placement", + "mesh-delivery-assigner.views.wants" ], "invokes": [ "seat:mesh-hand-acts.hand-act", @@ -36,16 +42,16 @@ "mesh-delivery-view.node_haves", "seat:node-engine.manifest-reads", "mesh-delivery-indexer.indexer_config_readers", - "seat:mesh-vault.retire" + "seat:mesh-vault.retire", + "mesh-delivery-indexer.indexer_read_modules", + "mesh-delivery-indexer.indexer_read_manifests", + "mesh-delivery-indexer.indexer_read_predefined", + "mesh-delivery-indexer.indexer_read_config_candidates", + "mesh-delivery-assigner.assigner_read_placement", + "mesh-delivery-assigner.assigner_read_wants" ], "reads": [ - "mesh-delivery-assigner.placement", - "mesh-delivery-assigner.wants", - "mesh-delivery-builder.builds", - "mesh-delivery-indexer.modules", - "mesh-delivery-indexer.manifests", - "mesh-delivery-indexer.predefined", - "mesh-delivery-indexer.config-candidates" + "mesh-delivery-builder.builds" ], "tools": [ "config_declared", @@ -53,7 +59,9 @@ "config_renames", "config_rename_again", "config_rename_close", - "config_retiring" + "config_retiring", + "config_first_views", + "config_first_view_again" ], "state": [ "values", diff --git a/modules/mesh-vault/cmd/mesh-vault/main.go b/modules/mesh-vault/cmd/mesh-vault/main.go index 53db4aa2..b334d366 100644 --- a/modules/mesh-vault/cmd/mesh-vault/main.go +++ b/modules/mesh-vault/cmd/mesh-vault/main.go @@ -19,6 +19,8 @@ import ( "time" stdio "git.novox.be/novox/mesh-sdk/go" + + "git.novox.be/novox/mesh-delivery/sdk/views" ) func say(format string, args ...any) { @@ -108,21 +110,20 @@ func readHandAct(id string) (*HandActSeen, error) { return nil, nil } -// watchLeavers deletes a node's key of node-logins when its node key of the placement is deleted (noderotation.go): a -// watch, never a poll. It answers whether the watch began; the minute pass (LoginsPass) catches what it missed. +// watchLeavers begins the picture of the placement and deletes a node's key of node-logins when its node key of the +// placement is deleted (noderotation.go): the owner's event, never a poll. It answers whether it began; the minute pass +// (LoginsPass) catches what it missed. func watchLeavers(env *Env) bool { if env.Vault == nil { return true } - err := stdio.State(PlacementRead).Watch("*.node", func(c stdio.StateChange) error { - if c.Op == "delete" || c.Op == "purge" || len(c.Value) == 0 { - env.Vault.NodeLeft(strings.TrimSuffix(c.Key, ".node")) - } - return nil - }) - if err != nil { - say("the placement's nodes are not watched for one that leaves; the minute pass finds it: %v", err) - return false + if placementView == nil { + placementView = newPlacementView(stdio.Ask, func(key string) { + if node, ok := leaverKey(key); ok { + env.Vault.NodeLeft(node) + } + }) + beginPlacementView(placementView) } return true } @@ -232,17 +233,9 @@ func watchCredentialAsks(env *Env, kick chan<- struct{}) bool { return true } -// placement reads the seat mesh-placement's placement through the runtime under the vault's own account (its manifest reads it). -var placement = PlacementReader{ - Keys: func() ([]string, error) { return stdio.State(PlacementRead).Keys() }, - Read: func(key string) ([]byte, error) { - e, err := stdio.State(PlacementRead).Get(key) - if err != nil || e == nil { - return nil, err - } - return e.Value, nil - }, -} +// placement reads the seat mesh-placement's placement through its owner's tool and events (placementview.go), never the +// owner's bucket. +var placement = viewPlacement(func() *views.Follower { return placementView }) // readWant is one wanted account of the grants part, read afresh (issue-account). func readWant(user string) ([]byte, bool, error) { diff --git a/modules/mesh-vault/cmd/mesh-vault/mint_integration_test.go b/modules/mesh-vault/cmd/mesh-vault/mint_integration_test.go index 024f0b31..cad6ea00 100644 --- a/modules/mesh-vault/cmd/mesh-vault/mint_integration_test.go +++ b/modules/mesh-vault/cmd/mesh-vault/mint_integration_test.go @@ -63,6 +63,70 @@ type player struct { said []string // ask answers a mesh/ask (a call of another module's tool); nil answers none, as a call with no responder. ask func(key string, body json.RawMessage) (any, error) + + // The assigner's placement as its owner's tool and views family say it (ADR 0615): the values, and the revision of the + // last change, kept from the bucket. + placeMu sync.Mutex + placeValues map[string]json.RawMessage + placeRevision uint64 +} + +var placeCreated = time.Date(2026, 11, 4, 9, 0, 0, 0, time.UTC) + +// followPlacement keeps the bucket of the placement as the assigner's owner does: listed once, then each change said on +// its views family to the vault. Call it before the player serves. +func (p *player) followPlacement() { + p.t.Helper() + kv, err := p.js.KeyValue(p.ctx, "mesh-delivery-assigner_placement") + if err != nil { + p.t.Fatal(err) + } + w, err := kv.WatchAll(p.ctx) + if err != nil { + p.t.Fatal(err) + } + p.placeValues = map[string]json.RawMessage{} + listed := make(chan struct{}) + go func() { + current := true + for e := range w.Updates() { + if e == nil { + current = false + close(listed) + continue + } + p.placeMu.Lock() + deleted := e.Operation() != jetstream.KeyValuePut + if deleted { + delete(p.placeValues, e.Key()) + } else { + p.placeValues[e.Key()] = append(json.RawMessage(nil), e.Value()...) + } + if !current { + p.placeRevision++ + ev := map[string]any{"v": 1, "owner": "mesh-delivery-assigner", "state": "placement", "key": e.Key(), + "revision": p.placeRevision, "created": placeCreated} + if deleted { + ev["deleted"] = true + } else { + ev["value"] = json.RawMessage(e.Value()) + } + body, _ := json.Marshal(ev) + p.mu.Lock() + p.next++ + id := fmt.Sprintf("o-%d", p.next) + p.mu.Unlock() + p.send(map[string]any{"id": id, "method": "mesh/event", "params": map[string]any{"envelope": map[string]any{ + "key": "mesh-delivery-assigner.views.placement", "node": "novox", "body": json.RawMessage(body)}}}) + } + p.placeMu.Unlock() + } + }() + select { + case <-listed: + case <-time.After(10 * time.Second): + p.t.Fatal("the placement was not listed") + } } // bucketOf is the bucket a state names: the vault's own by its local name, another's as `.`. @@ -224,6 +288,15 @@ func (p *player) handle(m wire) { Body json.RawMessage `json:"body"` } _ = json.Unmarshal(m.Params, &ask) + if ask.Key == PlacementTool && p.placeValues != nil { + p.placeMu.Lock() + answer := map[string]any{"v": 1, "owner": "mesh-delivery-assigner", "state": "placement", "revision": p.placeRevision, + "created": placeCreated, "values": p.placeValues, "at": time.Now()} + raw, _ := json.Marshal(answer) + p.placeMu.Unlock() + p.answer(m.ID, json.RawMessage(raw), nil) + return + } if p.ask != nil { result, err := p.ask(ask.Key, ask.Body) p.answer(m.ID, result, err) @@ -399,6 +472,7 @@ func TestTheVaultWholeMintsOnTheBusWhoeverHoldsTheLease(t *testing.T) { {ID: "act-ok", By: "novox.mesh-config", At: time.Now(), Verb: "mesh-vault.retire", Args: []string{"g14 dunst token ret-1"}}, {ID: "act-other", By: "novox.mesh-config", At: time.Now(), Verb: "mesh-vault.retire", Args: []string{"g14 dunst other ret-1"}}}}, nil } + p.followPlacement() go p.serve(stdout) status := func() MintState { diff --git a/modules/mesh-vault/cmd/mesh-vault/placementview.go b/modules/mesh-vault/cmd/mesh-vault/placementview.go new file mode 100644 index 00000000..d0a06eeb --- /dev/null +++ b/modules/mesh-vault/cmd/mesh-vault/placementview.go @@ -0,0 +1,182 @@ +package main + +// The vault's reading of the seat mesh-placement's placement, through its owner (novox/hq ADR 0615 §1, §4 to §6, ADR +// 0596): the assigner's tool `assigner_read_placement` asked for the whole state at start, the owner's `views.placement` +// events after it, and the tool once more only on a gap. Started while the assigner is down it keeps asking and says so; +// it never reads the assigner's bucket instead. (The vault's other reads, the node-engines' haves, the grants part's +// accounts and the provisioner's plans and provisions, are not the indexer's or the assigner's state and stay as they are.) + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" + + "git.novox.be/novox/mesh-delivery/sdk/views" +) + +// PlacementTool is the assigner's ordinary read tool of its placement; PlacementViewsEvent the event of its views family. +const ( + PlacementTool = delivery.AssignerModule + ".assigner_read_placement" + PlacementViewsEvent = delivery.AssignerModule + ".views.placement" + FirstViewsTool = "vault_first_views" + FirstViewAgainTool = "vault_first_view_again" +) + +// placementView is the picture; nil until main begins it. +var placementView *views.Follower + +// newPlacementView makes the follower: ask calls the tool, gone is told of a deleted key. +func newPlacementView(ask func(key string, body any) (json.RawMessage, error), gone func(key string)) *views.Follower { + return &views.Follower{Owner: delivery.AssignerModule, Tool: PlacementTool, State: "placement", Logf: say, + Ask: func(_ context.Context, after string) (views.Answer, error) { + body := map[string]any{"values": true} + if after != "" { + body["after"] = after + } + raw, err := ask(PlacementTool, body) + if err != nil { + return views.Answer{}, err + } + return views.DecodeAnswer(unwrapToolAnswer(raw)) + }, + Raise: func(s views.Status) { + say("the vault has no first view of the assigner's placement: %s has not answered since %s (%s); it keeps asking", s.Tool, + s.Asked.Format(time.RFC3339), s.Error) + }, + Changed: func(key string, _ json.RawMessage, deleted bool) { + if deleted { + gone(key) + } + }} +} + +// unwrapToolAnswer takes a tool's answer out of the protocol's content list when it comes in one. +func unwrapToolAnswer(raw json.RawMessage) []byte { + var r struct { + Content []struct { + Text string `json:"text"` + } `json:"content"` + } + if json.Unmarshal(raw, &r) == nil && len(r.Content) == 1 { + return []byte(r.Content[0].Text) + } + return raw +} + +// beginPlacementView asks the first view until answered and hears the owner's events. +func beginPlacementView(f *views.Follower) { + go func() { + if err := f.Start(context.Background()); err != nil { + say("the first view of the assigner's placement ended unanswered: %v", err) + } + }() + go func() { + <-stdio.Served() + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + err := stdio.Subscribe(PlacementViewsEvent, func(e stdio.Envelope) error { + ev, err := views.DecodeEvent(e.Body) + if err != nil { + say("%s is not a views event: %v", PlacementViewsEvent, err) + return nil + } + if gap, why := f.Apply(ev); gap { + say("a gap in the assigner's placement (%s): its first view is asked once more", why) + go func() { _ = f.Start(context.Background()) }() + } + return nil + }) + if err == nil { + return + } + say("does not hear %s yet (%v); asking again in %s", PlacementViewsEvent, err, wait) + time.Sleep(wait) + } + }() +} + +// viewPlacement is the placement reader over the picture: a key asked before the first view is answered is an error, so +// the caller waits (the "not yet" of the vault's other reads), never a read of the owner's bucket. +func viewPlacement(f func() *views.Follower) PlacementReader { + ready := func() (*views.Follower, error) { + fol := f() + if fol == nil { + return nil, fmt.Errorf("the vault reads the assigner's placement only when it runs as the module") + } + select { + case <-fol.Ready(): + return fol, nil + default: + s := fol.Status() + return nil, fmt.Errorf("the first view of the assigner's placement is not answered yet (%s asked %d times)", s.Tool, s.Asks) + } + } + return PlacementReader{ + Keys: func() ([]string, error) { + fol, err := ready() + if err != nil { + return nil, err + } + var keys []string + for k := range fol.Values() { + keys = append(keys, k) + } + sort.Strings(keys) + return keys, nil + }, + Read: func(key string) ([]byte, error) { + fol, err := ready() + if err != nil { + return nil, err + } + v, ok := fol.Get(key) + if !ok { + return nil, nil + } + return v, nil + }, + } +} + +// leaverKey is whether a deleted key of the placement is a node's own key, `.node`. +func leaverKey(key string) (string, bool) { + node, ok := strings.CutSuffix(key, ".node") + return node, ok && node != "" +} + +// viewTools are the read tool of the first view and its re-read, as every reader has them (design 25 § 12.2). +func viewTools(env *Env) []toolDef { + return []toolDef{ + {Name: FirstViewsTool, + Description: "The vault's first view of the assigner's placement: the tool it asks, the revision its picture holds, when it " + + "asked and was answered, how many asks (and by hand), and whether it is waiting.", + run: func(string, map[string]any) (any, error) { + if placementView == nil { + return nil, fmt.Errorf("the vault reads the assigner's placement only when it runs as the module") + } + return map[string]any{"first-views": []views.Status{placementView.Status()}}, nil + }}, + {Name: FirstViewAgainTool, + Description: "Ask the first view of the assigner's placement once more by hand, with why, recorded in the vault's log.", + Input: map[string]any{"why": str("why the first view is asked again")}, + run: func(caller string, a map[string]any) (any, error) { + if placementView == nil { + return nil, fmt.Errorf("the vault reads the assigner's placement only when it runs as the module") + } + why := strings.TrimSpace(arg(a, "why")) + if why == "" { + return nil, fmt.Errorf("say why: a first view asked by hand keeps its why") + } + if err := fromAGrantedCaller(caller, FirstViewAgainTool); err != nil { + return nil, err + } + say("the first view of the assigner's placement is asked again by %s: %s", caller, why) + return map[string]any{"asked-again": true}, placementView.Again(context.Background(), why) + }}, + } +} diff --git a/modules/mesh-vault/cmd/mesh-vault/rotation_integration_test.go b/modules/mesh-vault/cmd/mesh-vault/rotation_integration_test.go index 2db1d205..fcb44b18 100644 --- a/modules/mesh-vault/cmd/mesh-vault/rotation_integration_test.go +++ b/modules/mesh-vault/cmd/mesh-vault/rotation_integration_test.go @@ -88,6 +88,7 @@ func TestTheVaultWholeRotatesANodesLoginWithAStandInEngine(t *testing.T) { } return map[string]any{"acts": []HandActSeen{{ID: "act-1", By: "person.jochen", At: time.Now(), Verb: "rotate-node", Args: []string{"g14"}}}}, nil } + p.followPlacement() go p.serve(stdout) // The bus node (novox, running nats) and the node that rotates (g14); no lease key anywhere (issue 1126: the controller diff --git a/modules/mesh-vault/cmd/mesh-vault/tools.go b/modules/mesh-vault/cmd/mesh-vault/tools.go index f15e428b..cc5d4b19 100644 --- a/modules/mesh-vault/cmd/mesh-vault/tools.go +++ b/modules/mesh-vault/cmd/mesh-vault/tools.go @@ -79,7 +79,7 @@ func str(description string) map[string]any { } func definitions(env *Env) []toolDef { - return append(enrolmentTools(env), []toolDef{ + return append(append(enrolmentTools(env), viewTools(env)...), []toolDef{ {Name: Seat + ".get", Description: "One secret, sealed to the public key the asker sends (the sealed box of ADR 0206), with its " + "generation; answered only to an asker the vault allows, as the runtime proved it (ADR 0349 §11). " + diff --git a/modules/mesh-vault/cmd/mesh-vault/vault_test.go b/modules/mesh-vault/cmd/mesh-vault/vault_test.go index fd113fba..f06e1b9d 100644 --- a/modules/mesh-vault/cmd/mesh-vault/vault_test.go +++ b/modules/mesh-vault/cmd/mesh-vault/vault_test.go @@ -151,9 +151,9 @@ func TestNoBucketHoldsASecret(t *testing.T) { "buckets are the accounts' hashes (ADR 0547 clause 9), the signing keys' public halves (ADR 0550 §2) and the "+ "credentials' outcomes, names and generations alone (ADR 0570 clauses 3, 6) and each node's login in use (ADR 0583)", s) } - if r, _ := json.Marshal(manifest["reads"]); string(r) != `["node-engine.have","mesh-grants.accounts","mesh-delivery-assigner.placement","mesh-provisioner.plans","mesh-provisioner.provisions"]` { + if r, _ := json.Marshal(manifest["reads"]); string(r) != `["node-engine.have","mesh-grants.accounts","mesh-provisioner.plans","mesh-provisioner.provisions"]` { t.Errorf("the manifest reads %s: the vault reads the seat mesh-provisioner's plans and provisions, the haves, the "+ - "grants part's wanted accounts and the placement alone, never the controller's lease (issue 1126)", r) + "grants part's wanted accounts alone (the placement is the assigner's, through its tool and events, ADR 0615), never the controller's lease (issue 1126)", r) } sources, _ := filepath.Glob("*.go") bucket := regexp.MustCompile(`stdio\.State\(|nats-io|jetstream|KeyValue`) diff --git a/modules/mesh-vault/go.mod b/modules/mesh-vault/go.mod index 86199626..85849672 100644 --- a/modules/mesh-vault/go.mod +++ b/modules/mesh-vault/go.mod @@ -3,6 +3,7 @@ module mesh-vault go 1.26.0 require ( + git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103174208-b7abe203a929 github.com/nats-io/nats-server/v2 v2.11.17 diff --git a/modules/mesh-vault/module.json b/modules/mesh-vault/module.json index edf4f98b..0a575213 100644 --- a/modules/mesh-vault/module.json +++ b/modules/mesh-vault/module.json @@ -28,12 +28,12 @@ "own-retired" ], "invokes": [ - "seat:mesh-hand-acts.hand-acts" + "seat:mesh-hand-acts.hand-acts", + "mesh-delivery-assigner.assigner_read_placement" ], "reads": [ "node-engine.have", "mesh-grants.accounts", - "mesh-delivery-assigner.placement", "mesh-provisioner.plans", "mesh-provisioner.provisions" ], @@ -211,7 +211,9 @@ "vault_signers", "vault_signed", "signers_reassert", - "node_logins" + "node_logins", + "vault_first_views", + "vault_first_view_again" ], "claims": [ { @@ -245,5 +247,8 @@ "mesh-config" ] } - } + }, + "consumes": [ + "mesh-delivery-assigner.views.placement" + ] }