Feature 57 (config and provisioning): release mesh-tools (the provider's StateBus, the SDK pin, templates), reviewed
mesh/merge-gate head 33d136e70227 merged into trunk 6b3e15042fee — pass, with a note: builds mesh-tools, node-tools (+40 dependent(s)) → ace, g14, novox, s…
mesh/repo-check head 33d136e70227 merged into trunk 6b3e15042fee — pass: THE CHANGE ALTERS ITS OWN CHECK (merge-check.sh): main's version judged it; the ch…
mesh/delivery delivered

This commit is contained in:
2026-10-16 09:41:17 +00:00
parent 6b3e15042f
commit 33d136e702
14 changed files with 963 additions and 8 deletions
+98
View File
@@ -0,0 +1,98 @@
package bus
import (
"encoding/json"
"errors"
"fmt"
"strings"
"time"
"github.com/nats-io/nats.go"
)
// A module's code asks its machine's provisioner client for its values at runtime (novox/hq ADR 0349 clause 14 as
// ADR 0370 changes it, issue 685).
//
// While a module's code runs here, its asks reach the bus under this runtime's account, and the bus proves only
// that account. So the runtime asks its own machine's client on ProvisionAskSubject — which the controller grants
// this machine's tool runner alone to publish (ADR 0370 §3) — naming the module itself, from the bundle that asks
// and never from the module's code or arguments (§4). The client answers that module's config values, for what its
// manifest declares, and no secret: the vault hands a secret only to a caller the bus proves (§2).
// ProvisionAskSubject is where this machine's tool runner asks its provisioner client (mesh-controller
// internal/broker ProvisionerAskSubject): `prov.ask.<node>`.
func ProvisionAskSubject(node string) string { return "prov.ask." + node }
// ProvisionChangedSubjects is every module's word of what moved on a machine (mesh-controller internal/broker
// ProvisionerChangedSubject): `prov.changed.<node>.<module>`, a module's name dots and all.
func ProvisionChangedSubjects(node string) string { return "prov.changed." + node + ".>" }
// ProvisionTimeout bounds one ask of the client; a variable so a test need not wait.
var ProvisionTimeout = 10 * time.Second
// ProvisionChanged is what the client says moved of one module: the kind, and the key, or every key of the kind when
// empty. Never the value.
type ProvisionChanged struct {
Kind string `json:"kind"`
Key string `json:"key,omitempty"`
}
// ProvisionRefused is the client's refusal, in its own words: a key the module does not declare, a module not placed
// on the machine, a secret, a kind it does not answer yet.
type ProvisionRefused struct{ Reason string }
func (r *ProvisionRefused) Error() string { return r.Reason }
// AskProvisioner asks this machine's provisioner client for one value of module: kind `config` or `bound`, and the
// key. The module is the caller's to name from the bundle it runs; this never takes it from what a bundle sent.
func (c *Conn) AskProvisioner(module, kind, key string) (json.RawMessage, error) {
if c.node == "" {
return nil, errors.New("this runtime names no machine, so it has no provisioner client to ask")
}
body, err := json.Marshal(map[string]string{"module": module, "kind": kind, "key": key})
if err != nil {
return nil, err
}
msg, err := c.nc.Request(ProvisionAskSubject(c.node), body, ProvisionTimeout)
if errors.Is(err, nats.ErrNoResponders) {
return nil, fmt.Errorf("%s's provisioner client does not answer: it is not running there, or the bus does "+
"not let %s ask it", c.node, c.cred.User)
}
if err != nil {
return nil, fmt.Errorf("asking %s's provisioner client for %s's %s %s: %w", c.node, module, kind, key, err)
}
var answered struct {
Value json.RawMessage `json:"value"`
Error string `json:"error"`
}
if err := json.Unmarshal(msg.Data, &answered); err != nil {
return nil, fmt.Errorf("%s's provisioner client answered something that is not its answer: %w", c.node, err)
}
if answered.Error != "" {
return nil, &ProvisionRefused{Reason: answered.Error}
}
if len(answered.Value) == 0 {
return nil, fmt.Errorf("%s's provisioner client answered no value and no reason", c.node)
}
return answered.Value, nil
}
// OnProvisionChanged hands told every word of this machine's provisioner client of what moved, by module, until
// stopped.
func (c *Conn) OnProvisionChanged(told func(module string, changed ProvisionChanged)) (func(), error) {
if c.node == "" {
return nil, errors.New("this runtime names no machine, so it hears no provisioner client")
}
prefix := "prov.changed." + c.node + "."
sub, err := c.nc.Subscribe(ProvisionChangedSubjects(c.node), func(m *nats.Msg) {
var changed ProvisionChanged
if json.Unmarshal(m.Data, &changed) != nil || changed.Kind == "" {
return
}
told(strings.TrimPrefix(m.Subject, prefix), changed)
})
if err != nil {
return nil, err
}
return func() { _ = sub.Unsubscribe() }, c.nc.Flush()
}
@@ -44,6 +44,10 @@ func (b *subscribing) Watch(json.RawMessage, func(json.RawMessage) error) (func(
func (b *subscribing) Seat(string, json.RawMessage, func(string, any) (json.RawMessage, error)) (json.RawMessage, error) {
return nil, nil
}
func (b *subscribing) Provision(string, json.RawMessage) (json.RawMessage, error) { return nil, nil }
func (b *subscribing) ProvisionWatch(json.RawMessage, func(json.RawMessage) error) (func(), error) {
return func() {}, nil
}
func (b *subscribing) Subscribe(d func(json.RawMessage) error) error {
b.mu.Lock()
defer b.mu.Unlock()
+29
View File
@@ -83,6 +83,11 @@ type Bus interface {
// `take` and `answer` bind once and hand each message or proof on through handOn, to whichever child
// of the module asked last — a restarted child asks again as its code runs again.
Seat(verb string, params json.RawMessage, handOn func(method string, params any) (json.RawMessage, error)) (json.RawMessage, error)
// Provision answers a bundle's `get` of one of its module's values at runtime, and ProvisionWatch hands deliver
// that value now and each change after, as the module whose bundle asks and never a module the bundle names
// (novox/hq ADR 0349 §14, ADR 0370 §4): its config, and its bindings' values that are not secret.
Provision(verb string, params json.RawMessage) (json.RawMessage, error)
ProvisionWatch(params json.RawMessage, deliver func(change json.RawMessage) error) (stop func(), err error)
}
// EventTimeout bounds how long a bundle has to handle one event before it is offered again.
@@ -307,6 +312,30 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an
if answered, err = mesh.Log(strings.TrimPrefix(m.Method, "mesh/log."), m.Params); err == nil {
result = answered
}
case "mesh/provision.get":
var answered json.RawMessage
if answered, err = mesh.Provision("get", m.Params); err == nil {
result = answered
}
case "mesh/provision.watch":
// As a state's watch: each change asked of this child in order, naming the watch by the
// id the child asked it with, and answered once the current value is handed over.
var stop func()
watch := m.ID
stop, err = mesh.ProvisionWatch(m.Params, func(change json.RawMessage) error {
_, err := c.ask(module, "mesh/provision", map[string]any{"watch": watch, "change": change}, EventTimeout)
return err
})
if err == nil {
c.mu.Lock()
if c.why != nil {
c.mu.Unlock()
stop()
} else {
c.watches = append(c.watches, stop)
c.mu.Unlock()
}
}
case "mesh/state.watch":
// Each change is asked of this child, in order, naming the watch it is for by
// the id the child asked it with — two watches of one state are two handlers;
@@ -0,0 +1,107 @@
package launch
import (
"encoding/json"
"os"
"path/filepath"
"slices"
"strings"
"sync"
"testing"
"time"
)
// A bundle that asks for one of its values and watches it, and writes each change it is handed to $OUT.
const provisionBundle = `#!/bin/sh
while IFS= read -r line; do
id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\)[,}].*/\1/p')
case "$line" in
*'"method":"initialize"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;;
*'"method":"tools/list"'*)
printf '{"jsonrpc":"2.0","id":%s,"result":{"tools":[]}}\n' "$id"
printf '{"jsonrpc":"2.0","id":"g1","method":"mesh/provision.get","params":{"kind":"config","key":"role"}}\n'
printf '{"jsonrpc":"2.0","id":"w1","method":"mesh/provision.watch","params":{"kind":"config","key":"role"}}\n' ;;
*'"method":"mesh/provision"'*)
printf '%s\n' "$line" >> "$OUT"
printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;;
*'"id":"g1"'*) printf '%s\n' "$line" >> "$OUT" ;;
esac
done
`
type provisioning struct {
subscribing
mu sync.Mutex
asked []string
}
func (b *provisioning) Provision(verb string, params json.RawMessage) (json.RawMessage, error) {
b.mu.Lock()
defer b.mu.Unlock()
b.asked = append(b.asked, verb+" "+string(params))
return json.RawMessage(`"the build machine"`), nil
}
func (b *provisioning) ProvisionWatch(params json.RawMessage, deliver func(json.RawMessage) error) (func(), error) {
b.mu.Lock()
b.asked = append(b.asked, "watch "+string(params))
b.mu.Unlock()
return func() {}, deliver(json.RawMessage(`{"kind":"config","key":"role","value":"the build machine"}`))
}
// novox/hq ADR 0349 §14, ADR 0370 §4: a bundle's ask for its values reaches the runtime's bus from its own channel,
// answered there, and a watch's change is handed to it naming the watch it asked.
func TestABundleAsksForItsValuesAndIsHandedItsWatchsChanges(t *testing.T) {
entry := filepath.Join(t.TempDir(), "bundle")
if err := os.WriteFile(entry, []byte(provisionBundle), 0o755); err != nil {
t.Fatal(err)
}
out := filepath.Join(t.TempDir(), "out")
b := &provisioning{}
l, err := Start("claude-code", entry, append(os.Environ(), "OUT="+out), b, t.Logf)
if err != nil {
t.Fatal(err)
}
t.Cleanup(l.Stop)
deadline := time.Now().Add(10 * time.Second)
for {
said, _ := os.ReadFile(out)
if strings.Count(string(said), "\n") >= 2 {
// Each line as the bundle read it, by what it is: the get's answer, and the watch's change.
got := map[string]string{}
for _, line := range strings.Split(strings.TrimSpace(string(said)), "\n") {
var m struct {
ID any `json:"id"`
Method string `json:"method"`
Result json.RawMessage `json:"result"`
Params json.RawMessage `json:"params"`
}
if err := json.Unmarshal([]byte(line), &m); err != nil {
t.Fatalf("the bundle read %s", line)
}
if m.Method != "" {
got[m.Method] = string(m.Params)
} else {
got["answer"] = string(m.Result)
}
}
if got["answer"] != `"the build machine"` {
t.Errorf("the get was answered %s", got["answer"])
}
if got["mesh/provision"] != `{"change":{"kind":"config","key":"role","value":"the build machine"},"watch":"w1"}` {
t.Errorf("the watch was handed %s", got["mesh/provision"])
}
break
}
if time.Now().After(deadline) {
t.Fatalf("the bundle was handed %q", said)
}
time.Sleep(20 * time.Millisecond)
}
b.mu.Lock()
defer b.mu.Unlock()
slices.Sort(b.asked)
if len(b.asked) != 2 || b.asked[0] != `get {"kind":"config","key":"role"}` || b.asked[1] != `watch {"kind":"config","key":"role"}` {
t.Errorf("the runtime's bus was asked %q", b.asked)
}
}
+132
View File
@@ -0,0 +1,132 @@
// Package provide is the runtime's half of node-tools being a provider of `mcp-endpoint`: it runs the mesh SDK's
// provider helper, so a consumer handed to the provisioner (claude-code's `${bound:mcp-endpoint:port}`) is answered
// from an entry node-tools writes in its own bucket, not from the controller's binding file (novox/hq ADR 0349
// clause 12, issue 791).
//
// The offer carries no credential: the console's authority is the machine's login (ADR 0152, 0175), so the helper
// reads nothing from the vault, needs no consumers' file and the entry names none. node-tools is not marked moved:
// its manifest says `"answered-by": "helper"` for the provision, and the helper runs whenever the runtime does
// (issue 791), as route-proxy's does (issue 792).
package provide
import (
"context"
"encoding/json"
"fmt"
"time"
"git.novox.be/novox/mesh-sdk/go/provider"
"github.com/novox/mesh-tools/node-tools/internal/bus"
)
const (
// Module is the provider: the runtime's own module.
Module = "node-tools"
// Provision is what node-tools provides, as its manifest says.
Provision = "mcp-endpoint"
// ServedPort is the port its manifest serves (`serves.mcp-endpoint.port`): where the console listens.
ServedPort = 4270
)
// Offers is what node-tools serves, per provision.
func Offers() map[string]provider.Offer {
return map[string]provider.Offer{Provision: {
Credential: provider.NoCredential,
Serves: map[string]any{"port": ServedPort},
}}
}
// busOf is the helper's bus over the runtime's own state: the provider's bucket is its state `bound`, and the
// controller's assignments are the state it reads, `mesh-controller.assignments` (both declared in the manifest).
type busOf struct{ conn *bus.Conn }
func (b busOf) state(bucket string) (string, error) {
switch bucket {
case provider.BucketOf(Module):
return "bound", nil
case provider.AssignmentsBucket:
return "mesh-controller.assignments", nil
}
return "", fmt.Errorf("%s is not a bucket this provider keeps or reads", bucket)
}
func (b busOf) Watch(ctx context.Context, bucket string, handle func(key string, value []byte)) error {
name, err := b.state(bucket)
if err != nil {
return err
}
stop, err := b.conn.StateWatch(Module, name, "", func(ch bus.StateChange) error {
if ch.Op == "delete" {
handle(ch.Key, nil)
} else {
handle(ch.Key, ch.Value)
}
return nil
})
if err != nil {
return err
}
go func() { <-ctx.Done(); stop() }()
return nil
}
func (b busOf) Put(_ context.Context, bucket, key string, value []byte) error {
name, err := b.state(bucket)
if err != nil {
return err
}
_, err = b.conn.StatePut(Module, name, key, json.RawMessage(value))
return err
}
func (b busOf) Delete(_ context.Context, bucket, key string) error {
name, err := b.state(bucket)
if err != nil {
return err
}
return b.conn.StateDelete(Module, name, key)
}
// New is the helper answering node-tools's consumers on node.
func New(conn *bus.Conn, node string, every time.Duration, logf func(string, ...any)) *provider.Provider {
return provider.New(provider.Options{
Node: node,
Module: Module,
Offers: Offers(),
Bus: busOf{conn: conn},
Every: every,
Logf: logf,
})
}
// Run answers the consumers until ctx ends. The state is not reachable until the mesh has issued the runtime its
// membership and made the buckets, so a start that cannot watch is tried again, not given up.
func Run(ctx context.Context, conn *bus.Conn, node string, every time.Duration, logf func(string, ...any)) {
for delay := 2 * time.Second; ctx.Err() == nil; delay = min(delay*2, time.Minute) {
attempt, cancel := context.WithCancel(ctx)
err := New(conn, node, every, logf).Run(attempt)
cancel()
if ctx.Err() != nil {
return
}
logf("[mesh-tools] the provider helper stopped: %v; trying again in %s", err, delay)
select {
case <-ctx.Done():
case <-time.After(delay):
}
}
}
// Start runs the helper when this process is the machine's runtime (the node-tools credential, on a machine's
// name), with no condition on the environment, and returns what ends it. A runtime on another credential, a module's
// own in a test, is not the provider and answers nobody.
func Start(conn *bus.Conn, module, node string, every time.Duration, logf func(string, ...any)) (stop func()) {
if module != Module || node == "" {
return func() {}
}
ctx, cancel := context.WithCancel(context.Background())
logf("[mesh-tools] answering %s's consumers of %s through the provider helper", Module, Provision)
go Run(ctx, conn, node, every, logf)
return cancel
}
+149
View File
@@ -0,0 +1,149 @@
package provide
import (
"encoding/json"
"os"
"strings"
"testing"
"time"
"github.com/novox/mesh-tools/node-tools/internal/bus"
mt "github.com/novox/mesh-tools/node-tools/internal/meshtest"
)
// claude-code's facts.json template, as the catalogue keeps it (modules/claude-code/files/facts.json).
const factsTemplate = `{
"node": "${machine:name}",
"console": "http://127.0.0.1:${bound:mcp-endpoint:port}/mcp"
}
`
// factsToday is that file as the controller composes it today, from the port node-tools serves.
const factsToday = `{
"node": "anchor",
"console": "http://127.0.0.1:4270/mcp"
}
`
// fill is what the consumer's client does with the template: the machine's name, and each ${bound:<provision>:<key>}
// from the provider's entry.
func fill(t *testing.T, template string, entry map[string]any) string {
t.Helper()
values, _ := entry["values"].(map[string]any)
out := strings.ReplaceAll(template, "${machine:name}", "anchor")
for key, v := range values {
raw, _ := json.Marshal(v)
out = strings.ReplaceAll(out, "${bound:mcp-endpoint:"+key+"}", string(raw))
}
return out
}
func TestTheRuntimeAnswersAHandedConsumersMcpEndpointThroughTheHelper(t *testing.T) {
mesh := mt.New(t)
bound := mesh.Bucket(t, "node-tools_bound")
assignments := mesh.Bucket(t, "mesh-controller_assignments")
mem := mt.MembershipOf("node-tools", "anchor", false, nil)
mem.State = []bus.StateIssued{
{Name: "bound", Bucket: "node-tools_bound", Writes: true},
{Name: "mesh-controller.assignments", Bucket: "mesh-controller_assignments"},
}
mesh.Issue(t, mem)
// claude-code, handed on this machine, is bound to node-tools here for mcp-endpoint (issue 788's plan entry).
plan := `{"node":"anchor","module":"claude-code","provisioning":{"provisions":[{"provision":"mcp-endpoint",` +
`"provider":"anchor","provider-module":"node-tools","at":"127.0.0.1","from":"anchor","as":"mesh_anchor_claude_code"}]}}`
if _, err := assignments.Put("anchor.claude-code", []byte(plan)); err != nil {
t.Fatal(err)
}
conn, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
if err != nil {
t.Fatal(err)
}
conn.Logf = func(string, ...any) {}
t.Cleanup(conn.Close)
conn.Follow("node-tools")
said := &mt.Logs{}
t.Cleanup(Start(conn, "node-tools", "anchor", 20*time.Millisecond, said.Logf))
var entry map[string]any
mt.Until(t, func() error {
kept, err := bound.Get("anchor.claude-code.mcp-endpoint")
if err != nil {
return err
}
return json.Unmarshal(kept.Value(), &entry)
})
values, _ := entry["values"].(map[string]any)
for key, want := range map[string]any{"port": float64(4270), "at": "127.0.0.1", "from": "anchor"} {
if values[key] != want {
t.Errorf("the entry's %s is %v, want %v (entry %v)", key, values[key], want, entry)
}
}
if entry["as"] != "mesh_anchor_claude_code" {
t.Errorf("the entry's as is %v", entry["as"])
}
if c, has := entry["credential"]; has {
t.Errorf("an offer with no credential names one: %v", c)
}
if got := fill(t, factsTemplate, entry); got != factsToday {
t.Errorf("claude-code's facts.json fills to\n%s\nwant today's\n%s", got, factsToday)
}
}
// The port the offer serves is the port the manifest serves and the console listens on.
func TestTheOfferServesThePortTheManifestServes(t *testing.T) {
raw, err := os.ReadFile("../../module.json")
if err != nil {
t.Fatal(err)
}
var m struct {
Serves map[string]map[string]any `json:"serves"`
Moved bool `json:"moved"`
Provides []struct {
Name string `json:"name"`
AnsweredBy string `json:"answered-by"`
} `json:"provides"`
Resources []struct {
ID string `json:"id"`
} `json:"resources"`
Build struct {
Artifacts []struct {
Name string `json:"name"`
Env map[string]string `json:"env"`
} `json:"artifacts"`
} `json:"build"`
}
if err := json.Unmarshal(raw, &m); err != nil {
t.Fatal(err)
}
if got := m.Serves[Provision]["port"]; got != float64(ServedPort) {
t.Errorf("the manifest serves %v for %s, the offer %d", got, Provision, ServedPort)
}
if m.Moved {
t.Error("node-tools is marked moved: that also hands the runtime's bus account to the vault (issue 791 round 1)")
}
if len(m.Provides) != 1 || m.Provides[0].Name != Provision || m.Provides[0].AnsweredBy != "helper" {
t.Errorf("node-tools does not say its %s is answered by the helper: %+v", Provision, m.Provides)
}
for _, r := range m.Resources {
if r.ID == "consumers" {
t.Error("the manifest still carries a consumers file the helper does not need")
}
}
for _, a := range m.Build.Artifacts {
if a.Name == "files" || len(a.Env) > 0 {
t.Errorf("the build still carries %+v, which only a moved module needs", a)
}
}
}
// Only the machine's runtime is the provider: a process on another credential, or with no machine, answers nobody.
func TestOnlyTheMachinesRuntimeStartsTheHelper(t *testing.T) {
said := &mt.Logs{}
for _, c := range []struct{ module, node string }{{"claude-code", "anchor"}, {"node-tools", ""}} {
Start(nil, c.module, c.node, time.Second, said.Logf)()
}
}
+216
View File
@@ -0,0 +1,216 @@
package runtime
import (
"bytes"
"encoding/json"
"errors"
"fmt"
"sync"
"github.com/novox/mesh-tools/node-tools/internal/bus"
)
// A module's code reads its values at runtime (novox/hq ADR 0349 clause 14 as ADR 0370 changes it, issue 685).
//
// A bundle asks `mesh/provision.get` and `mesh/provision.watch` with `{kind, key}`; the runtime asks its machine's
// provisioner client **as the module whose bundle asks** — the module this bus was made for, from the bundle it
// launched, never a name the bundle sent (ADR 0370 §4) — and hands the answer back. A watch hands the current value
// first and then each change: the client says which value moved, and the runtime asks again. A secret is refused
// here, with the reason, and never asked: the client serves no secret on the tool runner's word (§2), and a module
// under an account of its own asks the vault itself, which waits on issue 569.
// The kinds a bundle may ask.
const (
provisionConfig = "config"
provisionBound = "bound"
provisionSecret = "secret"
)
// provisionAsked is what a bundle names. Anything else it sends, a module's name among it, is not read.
type provisionAsked struct {
Kind string `json:"kind"`
Key string `json:"key"`
}
// ProvisionChange is what a watch hands a bundle: the value, or, when it is no longer answered, why.
type ProvisionChange struct {
Kind string `json:"kind"`
Key string `json:"key"`
Value json.RawMessage `json:"value,omitempty"`
// Refused is why the value is no longer answered: the module taken off the machine, the key no longer declared.
Refused string `json:"refused,omitempty"`
}
func readProvisionAsked(verb string, params json.RawMessage) (provisionAsked, error) {
var asked provisionAsked
if err := json.Unmarshal(params, &asked); err != nil || asked.Kind == "" || asked.Key == "" {
return asked, fmt.Errorf("mesh/provision.%s names no value: {kind, key}", verb)
}
switch asked.Kind {
case provisionConfig, provisionBound:
return asked, nil
case provisionSecret:
return asked, fmt.Errorf("the secret %s is not asked for at runtime while the module's code runs in the "+
"tool runner: the vault hands a secret only to a caller the bus proves, and the bus proves only the tool "+
"runner here (novox/hq ADR 0370 §2, ADR 0349 §11); the module's secrets reach it through its files, "+
"until it runs under a bus account of its own (issue 569)", asked.Key)
}
return asked, fmt.Errorf("mesh/provision.%s: %q is not a kind of value a module reads at runtime: config or bound",
verb, asked.Kind)
}
// Provision answers a bundle's `get` of one of its module's values.
func (b *moduleBus) Provision(verb string, params json.RawMessage) (json.RawMessage, error) {
if verb != "get" {
return nil, fmt.Errorf("the runtime answers no mesh/provision.%s", verb)
}
asked, err := readProvisionAsked(verb, params)
if err != nil {
return nil, err
}
return b.all.conn.AskProvisioner(b.module, asked.Kind, asked.Key)
}
// ProvisionWatch hands deliver the value now and each change after, until stopped. A value not answered now is
// refused now, with the client's reason.
func (b *moduleBus) ProvisionWatch(params json.RawMessage, deliver func(json.RawMessage) error) (func(), error) {
asked, err := readProvisionAsked("watch", params)
if err != nil {
return nil, err
}
value, err := b.all.conn.AskProvisioner(b.module, asked.Kind, asked.Key)
if err != nil {
return nil, err
}
w := &provisionWatch{bus: b, asked: asked, deliver: deliver, last: value}
if err := w.hand(ProvisionChange{Kind: asked.Kind, Key: asked.Key, Value: value}); err != nil {
return nil, err
}
stop, err := b.all.provisioned.add(b.all, b.module, w)
if err != nil {
return nil, err
}
return stop, nil
}
// provisionWatch is one watch a bundle holds: what it asked, and what it was handed last.
type provisionWatch struct {
bus *moduleBus
asked provisionAsked
deliver func(json.RawMessage) error
mu sync.Mutex // one change at a time, in order
last json.RawMessage
refused string
stopped bool
}
func (w *provisionWatch) hand(change ProvisionChange) error {
raw, err := json.Marshal(change)
if err != nil {
return err
}
return w.deliver(raw)
}
// again asks once more after the client said something moved, and hands the bundle what changed.
func (w *provisionWatch) again() {
w.mu.Lock()
defer w.mu.Unlock()
if w.stopped {
return
}
value, err := w.bus.all.conn.AskProvisioner(w.bus.module, w.asked.Kind, w.asked.Key)
change := ProvisionChange{Kind: w.asked.Kind, Key: w.asked.Key}
var refused *bus.ProvisionRefused
switch {
case errors.As(err, &refused):
if w.refused == refused.Reason {
return
}
change.Refused = refused.Reason
case err != nil:
// The client not reached: asked again on its next word, which it says whenever it comes up.
w.bus.all.logf("[mesh-tools] %s's watch of its %s %s could not ask again: %v", w.bus.module, w.asked.Kind,
w.asked.Key, err)
return
default:
if w.refused == "" && bytes.Equal(value, w.last) {
return
}
change.Value = value
}
if err := w.hand(change); err != nil {
w.bus.all.logf("[mesh-tools] %s did not take a change of its %s %s: %v", w.bus.module, w.asked.Kind,
w.asked.Key, err)
return
}
w.last, w.refused = value, change.Refused
}
// provisionWatches is every watch on this runtime, by module, and the one subscription that hears the client.
type provisionWatches struct {
mu sync.Mutex
of map[string]map[*provisionWatch]bool
stop func()
}
func (p *provisionWatches) add(all *consumers, module string, w *provisionWatch) (func(), error) {
p.mu.Lock()
defer p.mu.Unlock()
if p.stop == nil {
stop, err := all.conn.OnProvisionChanged(func(module string, changed bus.ProvisionChanged) {
for _, w := range p.told(module, changed) {
go w.again()
}
})
if err != nil {
return nil, fmt.Errorf("the runtime cannot hear its machine's provisioner client: %w", err)
}
p.stop = stop
}
if p.of == nil {
p.of = map[string]map[*provisionWatch]bool{}
}
if p.of[module] == nil {
p.of[module] = map[*provisionWatch]bool{}
}
p.of[module][w] = true
return func() {
w.mu.Lock()
w.stopped = true
w.mu.Unlock()
p.mu.Lock()
defer p.mu.Unlock()
delete(p.of[module], w)
}, nil
}
// told is the watches of module that what moved concerns.
func (p *provisionWatches) told(module string, changed bus.ProvisionChanged) []*provisionWatch {
p.mu.Lock()
defer p.mu.Unlock()
var out []*provisionWatch
for w := range p.of[module] {
if w.asked.Kind == changed.Kind && (changed.Key == "" || changed.Key == w.asked.Key) {
out = append(out, w)
}
}
return out
}
func (p *provisionWatches) stopAll() {
p.mu.Lock()
defer p.mu.Unlock()
for _, ws := range p.of {
for w := range ws {
w.mu.Lock()
w.stopped = true
w.mu.Unlock()
}
}
p.of = nil
if p.stop != nil {
p.stop()
p.stop = nil
}
}
@@ -0,0 +1,203 @@
package runtime
import (
"encoding/json"
"strings"
"sync"
"testing"
"time"
"github.com/nats-io/nats.go"
mt "github.com/novox/mesh-tools/node-tools/internal/meshtest"
)
// A module's code reads its values at runtime through moduleBus.Provision and ProvisionWatch (novox/hq ADR 0349
// §14, ADR 0370, issue 685), against a real bus with a stand-in for the machine's provisioner client: the runtime
// asks as the module whose bundle asks, whatever the bundle names; a watch hands the current value, then each change
// the client says moved, and a refusal once; a secret is refused without asking.
// standInClient answers prov.ask.<node> from its values, keyed `<module>.<kind>.<key>`, and keeps every ask.
type standInClient struct {
mu sync.Mutex
values map[string]string
asks []string
nc *nats.Conn
}
func aStandInClient(t *testing.T, node string, values map[string]string) *standInClient {
t.Helper()
nc, err := nats.Connect(mt.URL(t))
if err != nil {
t.Fatal(err)
}
t.Cleanup(nc.Close)
c := &standInClient{values: values, nc: nc}
if _, err := nc.Subscribe("prov.ask."+node, func(m *nats.Msg) {
var a struct{ Module, Kind, Key string }
_ = json.Unmarshal(m.Data, &a)
c.mu.Lock()
c.asks = append(c.asks, string(m.Data))
v, ok := c.values[a.Module+"."+a.Kind+"."+a.Key]
c.mu.Unlock()
if !ok {
_ = m.Respond([]byte(`{"error":"` + a.Module + ` declares no config value ` + a.Key + `"}`))
return
}
_ = m.Respond([]byte(`{"value":` + v + `}`))
}); err != nil {
t.Fatal(err)
}
_ = nc.Flush()
return c
}
func (c *standInClient) set(key, value string) {
c.mu.Lock()
defer c.mu.Unlock()
if value == "" {
delete(c.values, key)
return
}
c.values[key] = value
}
func (c *standInClient) asked() []string {
c.mu.Lock()
defer c.mu.Unlock()
return append([]string(nil), c.asks...)
}
func (c *standInClient) say(t *testing.T, node, module, body string) {
t.Helper()
if err := c.nc.Publish("prov.changed."+node+"."+module, []byte(body)); err != nil {
t.Fatal(err)
}
}
func TestABundleAsksForItsValuesAsItsOwnModule(t *testing.T) {
client := aStandInClient(t, "anchor", map[string]string{
"claude-code.config.managed_settings": `{"spinnerTipsEnabled":false}`,
"dunst.config.managed_settings": `"dunst's"`,
})
conn := connect(t, "node-tools", "anchor")
all := &consumers{conn: conn, logf: t.Logf, of: map[string]*moduleEvents{}}
b := all.forModule("claude-code").(*moduleBus)
// Its own value, though the bundle names another module: the runtime names the module from the bundle.
got, err := b.Provision("get", json.RawMessage(`{"kind":"config","key":"managed_settings","module":"dunst"}`))
if err != nil || string(got) != `{"spinnerTipsEnabled":false}` {
t.Fatalf("answered %s, %v", got, err)
}
if asks := client.asked(); len(asks) != 1 || asks[0] != `{"key":"managed_settings","kind":"config","module":"claude-code"}` {
t.Fatalf("the client was asked %v", asks)
}
// The client's refusal, in its words.
if _, err := b.Provision("get", json.RawMessage(`{"kind":"config","key":"nothing"}`)); err == nil ||
!strings.Contains(err.Error(), "declares no config value nothing") {
t.Errorf("an undeclared key: %v", err)
}
// A secret, refused here and never asked; and what is not an ask.
before := len(client.asked())
for params, want := range map[string]string{
`{"kind":"secret","key":"gitea-read"}`: "the vault hands a secret only to a caller the bus proves",
`{"kind":"setting","key":"role"}`: "not a kind of value",
`{"key":"role"}`: "names no value",
} {
if _, err := b.Provision("get", json.RawMessage(params)); err == nil || !strings.Contains(err.Error(), want) {
t.Errorf("%s: %v, want %q", params, err, want)
}
}
if _, err := b.ProvisionWatch(json.RawMessage(`{"kind":"secret","key":"gitea-read"}`), func(json.RawMessage) error {
return nil
}); err == nil {
t.Error("a secret was watched")
}
if n := len(client.asked()); n != before {
t.Errorf("the client was asked for what the runtime refuses: %v", client.asked()[before:])
}
}
func TestABundlesWatchIsHandedTheValueAndEachChange(t *testing.T) {
client := aStandInClient(t, "anchor", map[string]string{"claude-code.config.managed_settings": `{"a":1}`})
conn := connect(t, "node-tools", "anchor")
all := &consumers{conn: conn, logf: t.Logf, of: map[string]*moduleEvents{}}
t.Cleanup(all.stopAll)
b := all.forModule("claude-code").(*moduleBus)
handed := make(chan ProvisionChange, 16)
stop, err := b.ProvisionWatch(json.RawMessage(`{"kind":"config","key":"managed_settings","module":"dunst"}`),
func(raw json.RawMessage) error {
var c ProvisionChange
if err := json.Unmarshal(raw, &c); err != nil {
t.Errorf("handed %s", raw)
}
handed <- c
return nil
})
if err != nil {
t.Fatal(err)
}
next := func(what string) ProvisionChange {
t.Helper()
select {
case c := <-handed:
return c
case <-time.After(5 * time.Second):
t.Fatalf("nothing handed: %s", what)
}
return ProvisionChange{}
}
nothing := func(what string) {
t.Helper()
select {
case c := <-handed:
t.Errorf("%s: handed %+v", what, c)
case <-time.After(300 * time.Millisecond):
}
}
if c := next("the current value"); string(c.Value) != `{"a":1}` || c.Kind != "config" || c.Key != "managed_settings" {
t.Fatalf("first handed %+v", c)
}
// A change, said by the client: asked again, handed.
client.set("claude-code.config.managed_settings", `{"a":2}`)
client.say(t, "anchor", "claude-code", `{"kind":"config","key":"managed_settings"}`)
if c := next("the change"); string(c.Value) != `{"a":2}` {
t.Fatalf("after the change, handed %+v", c)
}
// Every key of the kind said to have moved, the value the same: nothing handed. Another module's or another
// key's word: not even asked.
client.say(t, "anchor", "claude-code", `{"kind":"config"}`)
nothing("the same value again")
before := len(client.asked())
client.say(t, "anchor", "dunst", `{"kind":"config","key":"managed_settings"}`)
client.say(t, "anchor", "claude-code", `{"kind":"config","key":"role"}`)
nothing("another's word")
if n := len(client.asked()); n != before {
t.Errorf("asked again for another's word: %v", client.asked()[before:])
}
// No longer answered: handed the refusal, once.
client.set("claude-code.config.managed_settings", "")
client.say(t, "anchor", "claude-code", `{"kind":"config"}`)
if c := next("the refusal"); c.Refused == "" || len(c.Value) != 0 {
t.Fatalf("handed %+v, want the refusal", c)
}
client.say(t, "anchor", "claude-code", `{"kind":"config"}`)
nothing("the same refusal again")
// Answered again, and then stopped: nothing after.
client.set("claude-code.config.managed_settings", `{"a":3}`)
client.say(t, "anchor", "claude-code", `{"kind":"config"}`)
if c := next("answered again"); string(c.Value) != `{"a":3}` {
t.Fatalf("handed %+v", c)
}
stop()
client.set("claude-code.config.managed_settings", `{"a":4}`)
client.say(t, "anchor", "claude-code", `{"kind":"config"}`)
nothing("a stopped watch")
for _, a := range client.asked() {
if !strings.Contains(a, `"module":"claude-code"`) {
t.Errorf("asked as another module: %s", a)
}
}
}
+3
View File
@@ -458,6 +458,8 @@ type consumers struct {
of map[string]*moduleEvents
// seatBound is each module's seat traffic bound once: a worker taken, a proof subject answered.
seatBound map[string]func()
// provisioned is every watch a module's code holds of its values at runtime (provision.go).
provisioned provisionWatches
}
type moduleEvents struct {
@@ -479,6 +481,7 @@ func (c *consumers) stopAll() {
stop()
}
c.seatBound = map[string]func(){}
c.provisioned.stopAll()
}
// forModule is the bus one launched bundle of a module reaches the mesh through.