novox/mesh-tools
18
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Read seat verbs' replaces from mesh-sdk's seat contract and seat holders from placement's seat-holders state, so the console asks the controller for neither (issue 1955, ADR 0571 clauses 4 and 5)

Browse Source 
Pins mesh-sdk f9a496d. The controller's nodes and modules reads remain.
This commit is contained in:
jschoubben
2026-10-29 23:13:42 +01:00
parent 9d67b0995f
commit 67850457ef
7 changed files with 133 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node-tools/go.mod
+1 -1
View File
@@ -8,7 +8,7 @@ require (
)
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214150-692dbf9a1791
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121
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
node-tools/go.sum
+2
View File
@@ -6,6 +6,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029213215-4a967933b368 h1:LaBxsyU3
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029213215-4a967933b368/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214150-692dbf9a1791 h1:HVKR7kENzltQVkfBN+2zhU9JOPZme7MjDkTvr/oiJ24=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214150-692dbf9a1791/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121 h1:OsA1v6n736a7RqEV8Nkip8pf5ILMbzj0PWuKqcBOhK0=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
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=
node-tools/internal/console/address.go
+73 -11
View File
@@ -17,6 +17,9 @@ package console
// session — so a tool that arrived a minute ago is found without the client reconnecting.
import (
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"context"
"encoding/json"
"fmt"
@@ -134,7 +137,11 @@ type index struct {
// what it serves in time, or it answered before and not now. An address it might answer is never
// called missing while it is here (2026-10-05).
Unheard []unheard
// Replaces is what each verb and tool replaces, from the controller's records (novox/hq ADR 0245):
// Placed is the state seat-holders as the seat mesh-placement's holder wrote it (novox/hq ADR 0571 clause 4), by its
// keys `<seat>` and `<seat>.<node>`, read with delivery.HoldersOf; PlacedWhy says why it could not be read.
Placed map[string][]byte
PlacedWhy string
// Replaces is what each verb and tool replaces (novox/hq ADR 0245):
// `seat:<seat>.<verb>` and `<module>.<tool>`.
Replaces map[string][]string
// Earlier is every runtime not heard this round whose last answer, within RuntimeForgotten, this
@@ -283,18 +290,13 @@ type recordedMachine struct {
func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery) []earlier) (*index, error) {
var wg sync.WaitGroup
var nodesOut, modulesOut string
var toolsAnswer json.RawMessage
var placed map[string][]byte
var placedErr error
wg.Add(3)
go func() { defer wg.Done(); nodesOut, _ = controllerOutput(conn, "nodes") }()
go func() { defer wg.Done(); modulesOut, _ = controllerOutput(conn, "modules") }()
// The seats' verbs as the records define them, for what each replaces (novox/hq ADR 0245): answered
// in-process from the records, never a command.
go func() {
defer wg.Done()
if got, err := conn.Ask("seat:mesh-controller.tools", map[string]any{}, ""); err == nil {
toolsAnswer = got.Result
}
}()
// Who holds which seat on which node, from placement (ADR 0571 clause 4): no call of the controller.
go func() { defer wg.Done(); placed, placedErr = seatHoldersOn(conn) }()
d, err := announce.Gather(conn, held)
wg.Wait()
if err != nil {
@@ -333,6 +335,10 @@ func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery
l := x.Listing
x.Discovery = d
x.Stale, x.Earlier = stale, earlierAt
x.Placed = placed
if placedErr != nil {
x.PlacedWhy = placedErr.Error()
}
for _, s := range d.Silent {
x.Unheard = append(x.Unheard, unheard{Runtime: s.Name, Machine: s.Machine,
Why: fmt.Sprintf("it answered PING; what it serves did not arrive within %s", announce.Patience)})
@@ -342,7 +348,7 @@ func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery
// a module with no tools is never a name here.
var recorded []recordedModule
recordedOK := json.Unmarshal([]byte(jsonIn(modulesOut)), &recorded) == nil
x.Replaces = replacesOf(toolsAnswer, recorded)
x.Replaces = replacesOf(catalogue.DefaultSeats(), recorded)
if recordedOK {
for _, m := range recorded {
if !m.Tools {
@@ -360,6 +366,10 @@ func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery
} else {
l.NotAnswering = append(l.NotAnswering, "mesh-controller (its records of the modules did not answer, so what is missing cannot be said)")
}
if x.PlacedWhy != "" {
l.NotAnswering = append(l.NotAnswering, delivery.SeatHoldersRead+" (who holds which seat by placement could not be read: "+
x.PlacedWhy+")")
}
for _, u := range x.Unheard {
l.NotAnswering = append(l.NotAnswering, "the runtime "+sayUnheard([]unheard{u}))
}
@@ -381,6 +391,50 @@ func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery
return x, nil
}
// seatHoldersOn reads the state seat-holders whole (novox/hq ADR 0571 clause 4): its current values, through the
// runtime's own membership, which reads it as delivery.SeatHoldersRead.
func seatHoldersOn(conn *bus.Conn) (map[string][]byte, error) {
var mu sync.Mutex
keys := map[string][]byte{}
stop, err := conn.StateWatch(conn.Credential().Module, delivery.SeatHoldersRead, "", func(c bus.StateChange) error {
if c.Current && c.Op == "put" {
mu.Lock()
keys[c.Key] = c.Value
mu.Unlock()
}
return nil
})
if err != nil {
return nil, err
}
stop()
mu.Lock()
defer mu.Unlock()
return keys, nil
}
// placedHolders is who holds seat (on node, for a seat held per machine) by placement, or none.
func (x *index) placedHolders(seat, node string) []delivery.SeatHolder {
if len(x.Placed) == 0 {
return nil
}
hs, err := delivery.HoldersOf(x.Placed, seat, node)
if err != nil || len(hs) > 0 || node != "" {
return hs
}
// A seat held per machine, asked without one: every node's key of it.
var all []delivery.SeatHolder
for key := range x.Placed {
if s, n, ok := strings.Cut(key, "."); ok && s == seat {
if more, err := delivery.HoldersOf(x.Placed, seat, n); err == nil {
all = append(all, more...)
}
}
}
sort.Slice(all, func(i, j int) bool { return all[i].Node+"/"+all[i].Module < all[j].Node+"/"+all[j].Module })
return all
}
func containsHolder(hs []holder, h holder) bool {
for _, x := range hs {
if x == h {
@@ -606,6 +660,14 @@ func resolve(x *index, address string) (target, error) {
return target{}, fmt.Errorf("%s is assigned to %s in the mesh's records, but nothing that answered discovery serves it: %s. "+
"Ask again in a moment; mesh_runtimes shows which runtimes answered", prefix, strings.Join(on, ", "), why)
}
if hs := x.placedHolders(prefix, node); len(hs) > 0 {
var held []string
for _, h := range hs {
held = append(held, h.Module+" on "+h.Node)
}
return target{}, fmt.Errorf("the seat %s is held by %s by placement, but no holder of it announced itself on the bus. "+
"Ask again in a moment; mesh_runtimes shows which runtimes answered", prefix, strings.Join(held, ", "))
}
if len(x.Unheard) > 0 {
return target{}, fmt.Errorf("nothing that answered discovery is called %s, but not every runtime answered: %s — "+
"%s may be theirs. Ask again in a moment; mesh_runtimes shows which runtimes answered", prefix, sayUnheard(x.Unheard), prefix)
node-tools/internal/console/runtimes_test.go
+37 -2
View File
@@ -2,10 +2,13 @@ package console
import (
"encoding/json"
"fmt"
"strings"
"sync/atomic"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go/micro"
"github.com/novox/mesh-tools/node-tools/internal/announce"
@@ -52,7 +55,19 @@ func TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed(t *testing.T)
// Kept for the test: one round of discovery answers every question but mesh_runtimes, which asks anew.
IndexKept, announce.Patience = time.Minute, 2500*time.Millisecond
t.Cleanup(func() { IndexKept, announce.Patience = keptWas, patienceWas })
mt.New(t)
mesh := mt.New(t)
// Who holds which seat, by placement (ADR 0571 clause 4), as the seat mesh-placement's holder composes it.
placedKV := mesh.Bucket(t, delivery.SeatHoldersBucket)
for key, sh := range delivery.SeatHoldersOf(map[string]delivery.Placement{"bench": {Node: "bench",
Holds: []delivery.Holding{{Claim: "mesh-wants", Scope: "mesh", Module: "mesh-delivery-assigner"}}}}) {
raw, _ := json.Marshal(sh)
if _, err := placedKV.Put(key, raw); err != nil {
t.Fatal(err)
}
}
consoleMembership := mt.MembershipOf("node-tools", "laptop-console", false, nil)
consoleMembership.State = []bus.StateIssued{{Name: delivery.SeatHoldersRead, Bucket: delivery.SeatHoldersBucket}}
mesh.Issue(t, consoleMembership)
laptop := connect(t, "node-tools", "laptop")
stopTool, err := laptop.HandleSubject("mesh.mod.slack.tool.slack_check.laptop", func(json.RawMessage) (any, error) {
@@ -77,13 +92,20 @@ func TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed(t *testing.T)
})
controller := connect(t, "mesh-controller", "bench")
var toolsAsked atomic.Int32
for verb, output := range map[string]string{
"nodes": `[{"name":"laptop"},{"name":"sleeper"},{"name":"desktop"},{"name":"bench"}]`,
"modules": `[{"module":"slack","on":["laptop"],"tools":true},{"module":"nap","on":["sleeper"],"tools":true},` +
`{"module":"nextcloud-client","on":["laptop","desktop","sleeper"],"tools":true}]`,
// No part keeps a central record of tools (ADR 0571 clause 5): a seat verb's `replaces` is read from its contract
// in mesh-sdk, so the controller's `tools` is served here only to see that it is never asked.
"tools": `{"seats":[]}`,
} {
output := output
output, verb := output, verb
stop, err := controller.HandleSubject("mesh.seat.mesh-controller.tool."+verb, func(json.RawMessage) (any, error) {
if verb == "tools" {
toolsAsked.Add(1)
}
return map[string]any{"output": output, "ok": true}, nil
})
if err != nil {
@@ -94,6 +116,13 @@ func TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed(t *testing.T)
controller.Flush()
console := connect(t, "node-tools", "laptop-console")
console.Follow("node-tools")
mt.Until(t, func() error {
if console.Membership("node-tools") == nil {
return fmt.Errorf("the console's membership is not issued yet")
}
return nil
})
up, err := Serve(NewSurface(console, "laptop.node-tools"), "127.0.0.1:0")
if err != nil {
t.Fatal(err)
@@ -113,8 +142,14 @@ func TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed(t *testing.T)
t.Errorf("the late runtime's machine lists nothing: %s", machine)
}
if n := toolsAsked.Load(); n != 0 {
t.Errorf("the console asked the controller's tools %d times", n)
}
// The silent one is named, never called missing.
for address, want := range map[string]string{
// A seat whose holder announced nothing is named with its holder by placement, never called missing.
"mesh-wants.read": "the seat mesh-wants is held by mesh-delivery-assigner on bench by placement",
"sleeper/nap.doze": "nap is assigned to sleeper in the mesh's records, but nothing that answered discovery serves it: discovery did not hear in full from node-tools on sleeper (it answered PING",
"sleeper/nextcloud-client.nextcloud_check": "nextcloud-client on sleeper did not answer discovery: discovery did not hear in full from node-tools on sleeper",
} {
node-tools/internal/console/search.go
+11 -21
View File
@@ -19,7 +19,8 @@ package console
// Ties go to a seat — the place tools belong — then to the shorter name, then by address.
import (
"encoding/json"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"path"
"sort"
"strings"
@@ -265,27 +266,16 @@ func (x *index) search(query string) []hit {
return hits
}
// replacesOf is what every verb and tool replaces, from the controller's records (novox/hq ADR 0245):
// a seat's verbs from its `tools` answer, keyed `seat:<seat>.<verb>`, and a module's own tools from
// its `modules` answer, keyed `<module>.<tool>`. A controller older than the field answers neither,
// and search matches names and descriptions as before.
func replacesOf(toolsAnswer json.RawMessage, modules []recordedModule) map[string][]string {
// replacesOf is what every verb and tool replaces (novox/hq ADR 0245): a seat's verbs from the seat's contract in
// mesh-sdk, keyed `seat:<seat>.<verb>`, read in-process and asked of no part (ADR 0571 clause 5: no part keeps a
// central record of tools); and a module's own tools from its records, keyed `<module>.<tool>`. A verb or a tool with
// no `replaces` is matched by its name and description as before.
func replacesOf(seats []catalogue.Seat, modules []recordedModule) map[string][]string {
out := map[string][]string{}
var seats struct {
Seats []struct {
Seat string `json:"seat"`
Tools []struct {
Name string `json:"name"`
Replaces []string `json:"replaces"`
} `json:"tools"`
} `json:"seats"`
}
if json.Unmarshal(toolsAnswer, &seats) == nil {
for _, s := range seats.Seats {
for _, t := range s.Tools {
if len(t.Replaces) > 0 {
out["seat:"+s.Seat+"."+t.Name] = t.Replaces
}
for _, s := range seats {
for _, v := range s.Serves {
if len(v.Replaces) > 0 {
out["seat:"+s.Name+"."+v.Name] = append([]string(nil), v.Replaces...)
}
}
}
node-tools/internal/console/search_test.go
+7 -11
View File
@@ -1,7 +1,8 @@
package console
import (
"encoding/json"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"strings"
"testing"
@@ -19,6 +20,7 @@ func meshToSearch(t *testing.T) *index {
"status": "One unit as the service manager sees it now: its states, whether it starts at boot.",
"restart": "Restart one unit.",
"journal": "The last lines of one unit's journal (at most 2000).",
"failed": "The units that failed.",
}
for _, node := range []string{"anchor", "laptop"} {
for verb, does := range verbs {
@@ -44,13 +46,7 @@ func meshToSearch(t *testing.T) *index {
Description: does, Subject: "mesh.mod.docker.tool." + tool + ".anchor"})
}
x, _, _ := indexOfEndpoints(endpoints)
tools := json.RawMessage(`{"seats":[{"seat":"node-service-manager","scope":"node","tools":[
{"name":"units","replaces":["systemctl list-units","systemctl --failed"]},
{"name":"status","replaces":["systemctl status","systemctl is-active"]},
{"name":"restart","replaces":["systemctl restart"]},
{"name":"journal","replaces":["journalctl"]}]},
{"seat":"mesh-controller","scope":"mesh","tools":[{"name":"node","replaces":["hostnamectl","uptime"]},{"name":"nodes"}]}]}`)
x.Replaces = replacesOf(tools, []recordedModule{{Module: "docker", On: []string{"anchor"}, Tools: true,
x.Replaces = replacesOf(catalogue.DefaultSeats(), []recordedModule{{Module: "docker", On: []string{"anchor"}, Tools: true,
Replaces: map[string][]string{"docker_list": {"docker ps"}, "docker_logs": {"docker logs"}, "docker_restart": {"docker restart"}}}})
return x
}
@@ -68,7 +64,7 @@ func TestSearchFindsTheToolThatReplacesACommandFirst(t *testing.T) {
"systemctl status": "anchor/node-service-manager.status",
"systemctl status nats.service": "anchor/node-service-manager.status",
"systemctl restart mesh-controller": "anchor/node-service-manager.restart",
"systemctl --failed": "anchor/node-service-manager.units",
"systemctl --failed": "anchor/node-service-manager.failed",
"docker ps": "anchor/docker.docker_list",
"docker ps -a": "anchor/docker.docker_list",
"docker logs": "anchor/docker.docker_logs",
@@ -134,10 +130,10 @@ func TestSearchStillMatchesEveryWord(t *testing.T) {
}
}
// A controller older than `replaces` answers without it: search matches names and descriptions as before.
// A seat's contract and a module's records without `replaces`: search matches names and descriptions as before.
func TestSearchWithoutReplacesMatchesAsBefore(t *testing.T) {
x := meshToSearch(t)
x.Replaces = replacesOf(json.RawMessage(`{"seats":[{"seat":"node-service-manager","tools":[{"name":"journal"}]}]}`),
x.Replaces = replacesOf([]catalogue.Seat{{Name: "node-service-manager", Serves: []catalogue.Verb{{Name: "journal"}}}},
[]recordedModule{{Module: "docker"}})
if len(x.Replaces) != 0 {
t.Fatalf("replaces from records that have none: %v", x.Replaces)
node-tools/module.json
+2 -1
View File
@@ -66,6 +66,7 @@
],
"reads": [
"mesh-provisioner.provisions",
"mesh-provisioner.plans"
"mesh-provisioner.plans",
"mesh-delivery-assigner.seat-holders"
]
}