Merge pull request 'Feature 57 (config and provisioning): release mesh-tools' (#36) from release/57-config-and-provisioning into main
This commit was merged in pull request #36.
This commit is contained in:
+3
-2
@@ -37,8 +37,9 @@ if ! command -v node >/dev/null 2>&1 || [ ! -d node_modules/@novox/mesh-sdk ]; t
|
||||
# And what of the runtime launches no bundle: how a bundle's seat traffic reaches the bus, what its
|
||||
# environment holds, what a runtime on a module's own account serves (novox/hq ADR 0259), and that a write
|
||||
# by compare-and-set loses no update under contention (novox/hq research 042), and a module's log verbs
|
||||
# (novox/hq ADR 0297), and a key's history (novox/hq ADR 0307).
|
||||
runtimeless='^(TestALogAnswersEachVerbOnTheBus|TestALogPageIsBoundedByItsBytes|TestAnAppendAtAStaleExpectIsRefusedAsALostUpdate|TestALogNotIssuedIsRefusedWithTheReason|TestWhatALogCannotHoldIsRefusedWithTheReason|TestALogRefusesAValueNamingACredential|TestAKeysHistoryIsEveryValueKeptOldestFirst|TestAKeysHistoryIsRefusedToAModuleNotIssuedTheState|TestARuntimeOnAModulesOwnAccountServesThatModuleAlone|TestNoBusWordReachesABundle|TestAToolCallNeverReachesASeatsEventAcceptOrProof|TestABundlesSeatTrafficGoesThroughItsModulesBus|TestALostCompareAndSetCarriesItsCode|TestTwoWritersIncrementingOneCounterOnTheBusLoseNoUpdate|TestAStaleWriteIsRefusedAndWritesNothing|TestACallOnTheCallerNamedSubjectReachesTheModuleWithItsCaller|TestTheSubjectThatNamesNoCallerStillWorksAndNamesNone)$'
|
||||
# (novox/hq ADR 0297), and a key's history (novox/hq ADR 0307), and a module's values asked of its machine's
|
||||
# provisioner client (novox/hq ADR 0370).
|
||||
runtimeless='^(TestALogAnswersEachVerbOnTheBus|TestALogPageIsBoundedByItsBytes|TestAnAppendAtAStaleExpectIsRefusedAsALostUpdate|TestALogNotIssuedIsRefusedWithTheReason|TestWhatALogCannotHoldIsRefusedWithTheReason|TestALogRefusesAValueNamingACredential|TestAKeysHistoryIsEveryValueKeptOldestFirst|TestAKeysHistoryIsRefusedToAModuleNotIssuedTheState|TestARuntimeOnAModulesOwnAccountServesThatModuleAlone|TestNoBusWordReachesABundle|TestAToolCallNeverReachesASeatsEventAcceptOrProof|TestABundlesSeatTrafficGoesThroughItsModulesBus|TestALostCompareAndSetCarriesItsCode|TestTwoWritersIncrementingOneCounterOnTheBusLoseNoUpdate|TestAStaleWriteIsRefusedAndWritesNothing|TestACallOnTheCallerNamedSubjectReachesTheModuleWithItsCaller|TestTheSubjectThatNamesNoCallerStillWorksAndNamesNone|TestABundleAsksForItsValuesAsItsOwnModule|TestABundlesWatchIsHandedTheValueAndEachChange)$'
|
||||
fi
|
||||
if command -v gcc >/dev/null 2>&1; then
|
||||
CGO_ENABLED=1 go test -race -count=1 $packages
|
||||
|
||||
@@ -24,6 +24,7 @@ import (
|
||||
"github.com/novox/mesh-tools/node-tools/internal/alive"
|
||||
"github.com/novox/mesh-tools/node-tools/internal/bus"
|
||||
"github.com/novox/mesh-tools/node-tools/internal/console"
|
||||
"github.com/novox/mesh-tools/node-tools/internal/provide"
|
||||
"github.com/novox/mesh-tools/node-tools/internal/runtime"
|
||||
)
|
||||
|
||||
@@ -62,6 +63,10 @@ func main() {
|
||||
stopSaying = alive.Keep(conn, cred.Node, alive.Every, log.Printf)
|
||||
}
|
||||
|
||||
// As the machine's runtime it is also the provider of `mcp-endpoint`: it answers its consumers through the
|
||||
// SDK's helper wherever it runs, whether or not the module is handed (novox/hq ADR 0349 clause 12, issue 791).
|
||||
stopProviding := provide.Start(conn, cred.Module, cred.Node, time.Minute, log.Printf)
|
||||
|
||||
listen := os.Getenv("MESH_CONSOLE_LISTEN")
|
||||
if listen == "" && cred.Module == runtimeModule {
|
||||
listen = consoleListen
|
||||
@@ -84,6 +89,7 @@ func main() {
|
||||
signal.Notify(signals, syscall.SIGTERM, syscall.SIGINT)
|
||||
<-signals
|
||||
stopSaying()
|
||||
stopProviding()
|
||||
stop()
|
||||
if up != nil {
|
||||
_ = up.Close()
|
||||
|
||||
+2
-2
@@ -8,15 +8,15 @@ require (
|
||||
)
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261015210729-d3e182503c19
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect
|
||||
github.com/google/go-tpm v0.9.8 // indirect
|
||||
github.com/klauspost/compress v1.20.0 // indirect
|
||||
github.com/minio/highwayhash v1.0.4 // indirect
|
||||
github.com/nats-io/jwt/v2 v2.8.1 // indirect
|
||||
github.com/nats-io/nats-server/v2 v2.11.17 // indirect
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nkeys v0.4.16 // indirect
|
||||
github.com/nats-io/nuid v1.0.1 // indirect
|
||||
go.uber.org/automaxprocs v1.6.0 // indirect
|
||||
golang.org/x/crypto v0.57.0 // indirect
|
||||
golang.org/x/time v0.15.0 // indirect
|
||||
)
|
||||
|
||||
+2
-2
@@ -1,3 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261015210729-d3e182503c19 h1:mAMt3KFmrHzDiBv4MLL94opWcG9y73ISMrBUyisapQE=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261015210729-d3e182503c19/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
@@ -16,8 +18,6 @@ github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
|
||||
github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
|
||||
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
|
||||
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
|
||||
go.uber.org/automaxprocs v1.6.0 h1:O3y2/QNTOdbF+e/dpXNNW7Rx2hZ4sTIPyybbxyNqTUs=
|
||||
go.uber.org/automaxprocs v1.6.0/go.mod h1:ifeIMSnPZuznNm6jmdzmU3/bfk01Fe2fotchwEFJ8r8=
|
||||
golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
|
||||
golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
|
||||
golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)()
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
@@ -15,7 +15,8 @@
|
||||
"provides": [
|
||||
{
|
||||
"name": "mcp-endpoint",
|
||||
"scope": "node"
|
||||
"scope": "node",
|
||||
"answered-by": "helper"
|
||||
}
|
||||
],
|
||||
"serves": {
|
||||
@@ -59,5 +60,11 @@
|
||||
"binary": "node-tools"
|
||||
}
|
||||
]
|
||||
}
|
||||
},
|
||||
"state": [
|
||||
"bound"
|
||||
],
|
||||
"reads": [
|
||||
"mesh-controller.assignments"
|
||||
]
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user