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.<state> 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
This commit is contained in:
2026-11-04 12:08:56 +01:00
parent e1a182bf09
commit 692f18ef3b
15 changed files with 798 additions and 60 deletions
+40 -20
View File
@@ -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.<node>.
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 " +
@@ -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 {
@@ -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.<state>` 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 `<owner>.<state>` 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 ("<owner>.<state>", 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, `<owner>.<state>` (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": "<owner>.<state>, 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
}},
}
}
@@ -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))
}
}
@@ -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)
}
+1
View File
@@ -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
+18 -10
View File
@@ -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",
+15 -22
View File
@@ -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) {
@@ -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 `<owner>.<name>`.
@@ -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 {
@@ -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>.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)
}},
}
}
@@ -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
+1 -1
View File
@@ -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). " +
@@ -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`)
+1
View File
@@ -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
+9 -4
View File
@@ -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"
]
}