TLS-5, TLS-6: the tool runner holds the indexer's manifests and modules and the assigner's placement and seat-holders from their first views and views events, never their state, and the console gains mesh_first_views and mesh_first_view_again (ADR 0615, ADR 0621, issue 1966)
Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -14,6 +14,7 @@ package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
@@ -27,6 +28,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/firstview"
|
||||
"github.com/novox/mesh-tools/node-tools/internal/launch"
|
||||
"github.com/novox/mesh-tools/node-tools/internal/provide"
|
||||
"github.com/novox/mesh-tools/node-tools/internal/runtime"
|
||||
@@ -86,12 +88,16 @@ func main() {
|
||||
}
|
||||
var up *console.Listening
|
||||
if listen != "" {
|
||||
// The indexer's and the assigner's states the console indexes the mesh by come to it as first views, asked of the owners'
|
||||
// own tools and followed on their views families, never read from their state (novox/hq ADR 0615 §1).
|
||||
source, stopViews := firstview.Start(context.Background(), conn, log.Printf)
|
||||
defer stopViews()
|
||||
node := cred.Node
|
||||
if node == "" {
|
||||
node = "?"
|
||||
}
|
||||
who := node + "." + cred.Module
|
||||
up, err = console.Serve(console.NewSurface(conn, who), listen)
|
||||
up, err = console.Serve(console.NewSurface(conn, source, who), listen)
|
||||
if err != nil {
|
||||
log.Fatalf("node-tools: %v", err)
|
||||
}
|
||||
|
||||
@@ -8,6 +8,7 @@ require (
|
||||
)
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect
|
||||
github.com/google/go-tpm v0.9.8 // indirect
|
||||
|
||||
@@ -38,13 +38,15 @@ var IndexKept = 5 * time.Second
|
||||
|
||||
// The seven tools the console announces. Names of the API's kind — letters, digits, `_`, `-`.
|
||||
const (
|
||||
verbOverview = "mesh_overview"
|
||||
verbMachine = "mesh_machine"
|
||||
verbSearch = "mesh_search"
|
||||
verbDescribe = "mesh_describe"
|
||||
verbCall = "mesh_call"
|
||||
verbRuntimes = "mesh_runtimes"
|
||||
verbWait = "mesh_wait"
|
||||
verbOverview = "mesh_overview"
|
||||
verbMachine = "mesh_machine"
|
||||
verbSearch = "mesh_search"
|
||||
verbDescribe = "mesh_describe"
|
||||
verbCall = "mesh_call"
|
||||
verbRuntimes = "mesh_runtimes"
|
||||
verbFirstViews = "mesh_first_views"
|
||||
verbFirstAgain = "mesh_first_view_again"
|
||||
verbWait = "mesh_wait"
|
||||
)
|
||||
|
||||
// searchCap is how many matches a search answers before it says how many more there were.
|
||||
@@ -89,13 +91,22 @@ func discovery() []map[string]any {
|
||||
"arrive, its size in bytes, how many modules and tools it announced, whether it was shortened to fit the " +
|
||||
"bus, and when it was last heard — and every runtime or machine expected that was not heard. Read-only; " +
|
||||
"for when an address is said to be missing or on another machine."},
|
||||
{"name": verbFirstViews, "inputSchema": obj(map[string]any{}),
|
||||
"description": "The tool runner's first views of the four states it reads of the indexer and the assigner (the " +
|
||||
"assigner's placement and seat-holders, the indexer's manifests and modules): for each, the owner, the tool " +
|
||||
"asked, the revision held and when it was made, how many keys, when it was asked and answered, how often " +
|
||||
"it was asked and how often by hand, whether its condition is raised, and why it was last asked. Read-only."},
|
||||
{"name": verbFirstAgain, "inputSchema": obj(map[string]any{
|
||||
"why": map[string]any{"type": "string", "description": "why it is asked again, kept in the log"},
|
||||
}), "description": "Ask the tool runner's four first views again, now, and say what each answers. A hand act, recorded in " +
|
||||
"the runner's log with its reason; for when a picture is doubted, or after an owner's state was made again."},
|
||||
waitTool(),
|
||||
}
|
||||
}
|
||||
|
||||
func isDiscovery(name string) bool {
|
||||
switch name {
|
||||
case verbOverview, verbMachine, verbSearch, verbDescribe, verbCall, verbRuntimes, verbWait:
|
||||
case verbOverview, verbMachine, verbSearch, verbDescribe, verbCall, verbRuntimes, verbFirstViews, verbFirstAgain, verbWait:
|
||||
return true
|
||||
}
|
||||
return false
|
||||
@@ -248,7 +259,7 @@ type recordedModule struct {
|
||||
// `held` is what the console holds already, so an instance whose announcement is unchanged is not
|
||||
// asked for it again; `before` is the last answer of every runtime this round did not hear, which the
|
||||
// index holds instead, marked with its age (novox/hq issue 554).
|
||||
func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery) []earlier) (*index, error) {
|
||||
func indexOn(conn *bus.Conn, src Source, held announce.Known, before func(announce.Discovery) []earlier) (*index, error) {
|
||||
var wg sync.WaitGroup
|
||||
var placed map[string][]byte
|
||||
var placedErr error
|
||||
@@ -258,9 +269,9 @@ func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery
|
||||
wg.Add(2)
|
||||
// Which modules there are and where they are placed, and which machines the mesh has, from the bus (ADR 0571
|
||||
// clauses 4 and 5): the indexer's modules and manifests and the placement seat's placements, never any other record.
|
||||
go func() { defer wg.Done(); recorded, known, recordsWhy = recordsOn(conn) }()
|
||||
go func() { defer wg.Done(); recorded, known, recordsWhy = recordsOn(src) }()
|
||||
// Who holds which seat on which node, from placement (ADR 0571 clause 4): no other call.
|
||||
go func() { defer wg.Done(); placed, placedErr = seatHoldersOn(conn) }()
|
||||
go func() { defer wg.Done(); placed, placedErr = seatHoldersOn(src) }()
|
||||
d, err := announce.Gather(conn, held)
|
||||
wg.Wait()
|
||||
if err != nil {
|
||||
@@ -345,30 +356,19 @@ func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery
|
||||
return x, nil
|
||||
}
|
||||
|
||||
// stateOn reads a state the runtime's membership reads, whole: its current values by key, a deleted key absent.
|
||||
func stateOn(conn *bus.Conn, name string) (map[string][]byte, error) {
|
||||
var mu sync.Mutex
|
||||
keys := map[string][]byte{}
|
||||
stop, err := conn.StateWatch(conn.Credential().Module, name, "", 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
|
||||
// Source is where the console reads the four states of the indexer and the assigner it indexes the mesh by: the assigner's
|
||||
// placement and seat-holders, the indexer's modules and manifests, by `<owner>.<state>` (delivery.PlacementRead and the
|
||||
// rest). The tool runner's is its first views (firstview.Views), never the owners' state (novox/hq ADR 0615 §1).
|
||||
type Source interface {
|
||||
// State is every current value of a state by key; an error says why it cannot be told yet.
|
||||
State(name string) (map[string][]byte, error)
|
||||
// Entry is one value of a state, nil when it holds none.
|
||||
Entry(name, key string) ([]byte, error)
|
||||
}
|
||||
|
||||
// seatHoldersOn reads the state seat-holders whole (novox/hq ADR 0571 clause 4), as delivery.SeatHoldersRead.
|
||||
func seatHoldersOn(conn *bus.Conn) (map[string][]byte, error) {
|
||||
return stateOn(conn, delivery.SeatHoldersRead)
|
||||
func seatHoldersOn(src Source) (map[string][]byte, error) {
|
||||
return src.State(delivery.SeatHoldersRead)
|
||||
}
|
||||
|
||||
// The indexer's states the console reads (novox/hq ADR 0571 clauses 4 and 5): every module, and each manifest by its
|
||||
@@ -390,8 +390,8 @@ var manifestsKept sync.Map // digest → manifestFacts
|
||||
// recordsOn is every module the indexer holds, with where the placement seat places it and whether it declares tools,
|
||||
// and every machine placement names (delivery.ModuleNames, delivery.MachineNames). What cannot be read is said, one line
|
||||
// per state, for the listing's not answering.
|
||||
func recordsOn(conn *bus.Conn) (recorded []recordedModule, machines []string, why []string) {
|
||||
placements, err := stateOn(conn, delivery.PlacementRead)
|
||||
func recordsOn(src Source) (recorded []recordedModule, machines []string, why []string) {
|
||||
placements, err := src.State(delivery.PlacementRead)
|
||||
if err != nil {
|
||||
why = append(why, delivery.PlacementRead+" (where modules are placed could not be read, so what is missing cannot "+
|
||||
"be said: "+err.Error()+")")
|
||||
@@ -408,7 +408,7 @@ func recordsOn(conn *bus.Conn) (recorded []recordedModule, machines []string, wh
|
||||
on[m.Module] = append(on[m.Module], node)
|
||||
}
|
||||
}
|
||||
modules, err := stateOn(conn, modulesRead)
|
||||
modules, err := src.State(modulesRead)
|
||||
if err != nil {
|
||||
return nil, machines, append(why, modulesRead+" (which modules there are could not be read: "+err.Error()+")")
|
||||
}
|
||||
@@ -419,7 +419,7 @@ func recordsOn(conn *bus.Conn) (recorded []recordedModule, machines []string, wh
|
||||
for _, name := range names {
|
||||
r := recordedModule{Module: name, On: on[name]}
|
||||
if m, err := delivery.DecodeModule(modules[name]); err == nil && m.Manifest != "" {
|
||||
facts, err := manifestOf(conn, m.Manifest)
|
||||
facts, err := manifestOf(src, m.Manifest)
|
||||
if err != nil {
|
||||
why = append(why, name+" (its manifest could not be read, so whether it answers tools is not known: "+err.Error()+")")
|
||||
} else {
|
||||
@@ -432,18 +432,18 @@ func recordsOn(conn *bus.Conn) (recorded []recordedModule, machines []string, wh
|
||||
}
|
||||
|
||||
// manifestOf is what the console takes of the manifest of a digest, read from the indexer's manifests once.
|
||||
func manifestOf(conn *bus.Conn, digest string) (manifestFacts, error) {
|
||||
func manifestOf(src Source, digest string) (manifestFacts, error) {
|
||||
if kept, ok := manifestsKept.Load(digest); ok {
|
||||
return kept.(manifestFacts), nil
|
||||
}
|
||||
entry, err := conn.StateGet(conn.Credential().Module, manifestsRead, digest)
|
||||
entry, err := src.Entry(manifestsRead, digest)
|
||||
if err != nil {
|
||||
return manifestFacts{}, err
|
||||
}
|
||||
if entry == nil {
|
||||
return manifestFacts{}, fmt.Errorf("the indexer holds no manifest %s", digest)
|
||||
}
|
||||
m, err := delivery.DecodeManifest(entry.Value)
|
||||
m, err := delivery.DecodeManifest(entry)
|
||||
if err != nil {
|
||||
return manifestFacts{}, err
|
||||
}
|
||||
@@ -501,7 +501,7 @@ func (s *Surface) indexAsked(fresh bool) (*index, error) {
|
||||
return x, nil
|
||||
}
|
||||
s.mu.Unlock()
|
||||
x, err := indexOn(s.conn, s.held, s.before)
|
||||
x, err := indexOn(s.conn, s.src, s.held, s.before)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -926,6 +926,9 @@ func (s *Surface) discover(ctx context.Context, name string, args map[string]any
|
||||
if name == verbRuntimes {
|
||||
return s.runtimesAnswer()
|
||||
}
|
||||
if name == verbFirstViews || name == verbFirstAgain {
|
||||
return s.firstViewsAnswer(ctx, name, args)
|
||||
}
|
||||
if name == verbWait {
|
||||
return s.waitAnswer(ctx, args)
|
||||
}
|
||||
|
||||
@@ -106,7 +106,7 @@ func TestTheMeshsToolsAreFoundByAddress(t *testing.T) {
|
||||
controller.Flush()
|
||||
watcher.Flush()
|
||||
|
||||
up, err := Serve(NewSurface(nodeTools, "desk.node-tools"), "127.0.0.1:0")
|
||||
up, err := Serve(NewSurface(nodeTools, busSource{nodeTools}, "desk.node-tools"), "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -125,7 +125,7 @@ func TestTheMeshsToolsAreFoundByAddress(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
if got := strings.Join(names, ","); got != "mesh_overview,mesh_machine,mesh_search,mesh_describe,mesh_call,mesh_runtimes,mesh_wait" {
|
||||
if got := strings.Join(names, ","); got != "mesh_overview,mesh_machine,mesh_search,mesh_describe,mesh_call,mesh_runtimes,mesh_first_views,mesh_first_view_again,mesh_wait" {
|
||||
t.Errorf("announced %s", got)
|
||||
}
|
||||
|
||||
|
||||
@@ -63,8 +63,8 @@ func toolKey(name string, seats Seats) string {
|
||||
|
||||
// toolsOn is the flat catalogue (MESH_CONSOLE_FLAT=1): what announced itself on the bus, every
|
||||
// tool and seat verb, and the assignments with tools that did not (novox/hq ADR 0197).
|
||||
func toolsOn(conn *bus.Conn) (*Listing, error) {
|
||||
x, err := indexOn(conn, nil, nil)
|
||||
func toolsOn(conn *bus.Conn, src Source) (*Listing, error) {
|
||||
x, err := indexOn(conn, src, nil, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -71,7 +71,7 @@ func TestTheConsoleListsAndCallsOverHTTP(t *testing.T) {
|
||||
return nil
|
||||
})
|
||||
|
||||
flat := NewSurface(nodeTools, "desk.node-tools")
|
||||
flat := NewSurface(nodeTools, busSource{nodeTools}, "desk.node-tools")
|
||||
flat.Flat = true // the whole catalogue, as before ADR 0195: still reachable, no longer announced
|
||||
up, err := Serve(flat, "127.0.0.1:0")
|
||||
if err != nil {
|
||||
@@ -132,7 +132,7 @@ func TestTheConsoleListsAndCallsOverHTTP(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestTheConsoleListensOnLoopbackAndNowhereElse(t *testing.T) {
|
||||
if _, err := Serve(NewSurface(nil, "x"), "0.0.0.0:0"); err == nil || !strings.Contains(err.Error(), "loopback and nowhere else") {
|
||||
if _, err := Serve(NewSurface(nil, nil, "x"), "0.0.0.0:0"); err == nil || !strings.Contains(err.Error(), "loopback and nowhere else") {
|
||||
t.Errorf("a non-loopback console was not refused: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -72,7 +72,7 @@ func TestACallAfterARoundThatHeardNothingUsesWhatWasHeardBefore(t *testing.T) {
|
||||
}
|
||||
return n
|
||||
}
|
||||
surface := NewSurface(console, "laptop.node-tools")
|
||||
surface := NewSurface(console, busSource{console}, "laptop.node-tools")
|
||||
up, err := Serve(surface, "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
package console
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/sdk/views"
|
||||
)
|
||||
|
||||
// FirstViewer is the tool runner's four first views as the console shows them (novox/hq ADR 0615 §6; issue 1966 TLS-6): a
|
||||
// Source that also is one is asked by mesh_first_views and mesh_first_view_again.
|
||||
type FirstViewer interface {
|
||||
Status() []views.Status
|
||||
Again(ctx context.Context, why string) []error
|
||||
}
|
||||
|
||||
// firstViewsAnswer is mesh_first_views and mesh_first_view_again.
|
||||
func (s *Surface) firstViewsAnswer(ctx context.Context, name string, args map[string]any) map[string]any {
|
||||
fv, ok := s.src.(FirstViewer)
|
||||
if !ok {
|
||||
return failure("this tool runner holds no first views: it reads the indexer's and the assigner's states through " +
|
||||
"another source")
|
||||
}
|
||||
if name == verbFirstViews {
|
||||
return answerText(map[string]any{"first views": fv.Status()})
|
||||
}
|
||||
why, _ := args["why"].(string)
|
||||
if why = strings.TrimSpace(why); why == "" {
|
||||
why = "asked by hand with no reason given"
|
||||
}
|
||||
// A hand act, recorded in the runner's log with who asked and why (novox/hq ADR 0615 §6).
|
||||
log.Printf("[mesh-tools] hand act: %s asked the four first views again: %s", s.who, why)
|
||||
out := map[string]any{"asked again": why}
|
||||
if errs := fv.Again(ctx, why); len(errs) > 0 {
|
||||
var msgs []string
|
||||
for _, e := range errs {
|
||||
msgs = append(msgs, e.Error())
|
||||
}
|
||||
out["not answered"] = msgs
|
||||
}
|
||||
out["first views"] = fv.Status()
|
||||
return answerText(out)
|
||||
}
|
||||
@@ -0,0 +1,49 @@
|
||||
package console
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/sdk/views"
|
||||
)
|
||||
|
||||
// viewer is a Source that also holds first views.
|
||||
type viewer struct {
|
||||
busSource
|
||||
asked []string
|
||||
}
|
||||
|
||||
func (v *viewer) Status() []views.Status {
|
||||
return []views.Status{{Owner: "mesh-delivery-assigner", Tool: "placement", State: "placement", Revision: 4, Asks: 1}}
|
||||
}
|
||||
|
||||
func (v *viewer) Again(_ context.Context, why string) []error {
|
||||
v.asked = append(v.asked, why)
|
||||
return nil
|
||||
}
|
||||
|
||||
// issue 1966 TLS-6: the console's mesh_first_views shows each first view, mesh_first_view_again asks them again with its
|
||||
// reason, and a source holding none refuses both in words.
|
||||
func TestTheConsoleShowsAndAsksAgainTheRunnersFirstViews(t *testing.T) {
|
||||
text := func(r map[string]any) string {
|
||||
c, _ := json.Marshal(r["content"])
|
||||
return string(c)
|
||||
}
|
||||
v := &viewer{}
|
||||
s := &Surface{src: v, who: "desk.node-tools"}
|
||||
if got := text(s.discover(context.Background(), "mesh_first_views", nil)); !strings.Contains(got, "mesh-delivery-assigner") || !strings.Contains(got, "placement") {
|
||||
t.Errorf("mesh_first_views shows %s", got)
|
||||
}
|
||||
got := s.discover(context.Background(), "mesh_first_view_again", map[string]any{"why": "doubted"})
|
||||
if len(v.asked) != 1 || v.asked[0] != "doubted" || !strings.Contains(text(got), "asked again") {
|
||||
t.Errorf("mesh_first_view_again asked %v and answered %s", v.asked, text(got))
|
||||
}
|
||||
for _, name := range []string{"mesh_first_views", "mesh_first_view_again"} {
|
||||
none := &Surface{src: busSource{}, who: "desk.node-tools"}
|
||||
if r := none.discover(context.Background(), name, nil); r["isError"] != true || !strings.Contains(text(r), "holds no first views") {
|
||||
t.Errorf("%s without first views answers %s", name, text(r))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -35,8 +35,12 @@ func TestTheManifestNamesTheFirstFamiliesOneByOne(t *testing.T) {
|
||||
t.Errorf("a family named by a wildcard: %s", w)
|
||||
}
|
||||
}
|
||||
if len(m.Consumes) > 0 {
|
||||
t.Errorf("the tool runner consumes %v: a wait is no consumer (ADR 0367 §2)", m.Consumes)
|
||||
// A wait is no consumer (ADR 0367 §2): what it consumes is the four views families of its first views (ADR 0615 §5, issue
|
||||
// 1966 TLS-5), and nothing else.
|
||||
wantConsumes := []string{"mesh-delivery-assigner.views.placement", "mesh-delivery-assigner.views.seat-holders",
|
||||
"mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules"}
|
||||
if !reflect.DeepEqual(m.Consumes, wantConsumes) {
|
||||
t.Errorf("the tool runner consumes %v, want %v", m.Consumes, wantConsumes)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -47,6 +47,7 @@ type RPCError struct {
|
||||
// Surface answers MCP requests over one bus connection, as one account.
|
||||
type Surface struct {
|
||||
conn *bus.Conn
|
||||
src Source
|
||||
who string
|
||||
mu sync.Mutex
|
||||
known *Listing
|
||||
@@ -69,7 +70,7 @@ const MethodSeats = "mesh/seats"
|
||||
|
||||
// seatsAnswer is MethodSeats' answer: mesh-sdk's seat definitions, with each seat's holders by placement.
|
||||
func (s *Surface) seatsAnswer() map[string]any {
|
||||
placed, err := seatHoldersOn(s.conn)
|
||||
placed, err := seatHoldersOn(s.src)
|
||||
x := &index{Placed: placed}
|
||||
seats := []map[string]any{}
|
||||
for _, st := range catalogue.DefaultSeats() {
|
||||
@@ -91,8 +92,8 @@ func (s *Surface) seatsAnswer() map[string]any {
|
||||
}
|
||||
|
||||
// NewSurface is the surface over a connection, as `who`.
|
||||
func NewSurface(conn *bus.Conn, who string) *Surface {
|
||||
return &Surface{conn: conn, who: who, Flat: os.Getenv("MESH_CONSOLE_FLAT") == "1"}
|
||||
func NewSurface(conn *bus.Conn, src Source, who string) *Surface {
|
||||
return &Surface{conn: conn, src: src, who: who, Flat: os.Getenv("MESH_CONSOLE_FLAT") == "1"}
|
||||
}
|
||||
|
||||
func (s *Surface) listing() (*Listing, error) {
|
||||
@@ -103,7 +104,7 @@ func (s *Surface) listing() (*Listing, error) {
|
||||
return l, nil
|
||||
}
|
||||
s.mu.Unlock()
|
||||
l, err := toolsOn(s.conn)
|
||||
l, err := toolsOn(s.conn, s.src)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -123,7 +123,7 @@ func TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed(t *testing.T)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
up, err := Serve(NewSurface(console, "laptop.node-tools"), "127.0.0.1:0")
|
||||
up, err := Serve(NewSurface(console, busSource{console}, "laptop.node-tools"), "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,39 @@
|
||||
package console
|
||||
|
||||
import (
|
||||
"sync"
|
||||
|
||||
"github.com/novox/mesh-tools/node-tools/internal/bus"
|
||||
)
|
||||
|
||||
// busSource is a Source over a bus connection's own states, which a test's fixture mesh fills: the console's tests stand a
|
||||
// membership that reads the four states beside it, where the tool runner itself holds first views (firstview.Views).
|
||||
type busSource struct{ conn *bus.Conn }
|
||||
|
||||
func (b busSource) State(name string) (map[string][]byte, error) {
|
||||
var mu sync.Mutex
|
||||
keys := map[string][]byte{}
|
||||
stop, err := b.conn.StateWatch(b.conn.Credential().Module, name, "", 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
|
||||
}
|
||||
|
||||
func (b busSource) Entry(name, key string) ([]byte, error) {
|
||||
entry, err := b.conn.StateGet(b.conn.Credential().Module, name, key)
|
||||
if err != nil || entry == nil {
|
||||
return nil, err
|
||||
}
|
||||
return entry.Value, nil
|
||||
}
|
||||
@@ -33,7 +33,7 @@ func waitingRuntime(t *testing.T) (*Surface, *mt.Mesh) {
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return NewSurface(conn, "desk.node-tools"), mesh
|
||||
return NewSurface(conn, busSource{conn}, "desk.node-tools"), mesh
|
||||
}
|
||||
|
||||
// callWait is one tools/call of mesh_wait, answered as an agent host reads it.
|
||||
@@ -176,7 +176,7 @@ func TestMeshWaitRefusesAnEventOutsideTheGrantedFamilies(t *testing.T) {
|
||||
// A tool runner issued no families waits on nothing, and says so.
|
||||
func TestMeshWaitWithNoFamiliesIssuedWaitsOnNothing(t *testing.T) {
|
||||
mt.New(t)
|
||||
s := NewSurface(connect(t, "node-tools", "desk"), "desk.node-tools")
|
||||
s := NewSurface(connect(t, "node-tools", "desk"), nil, "desk.node-tools")
|
||||
text, isErr := callWait(s, map[string]any{"subjects": []any{"mesh-knowledge.reviewed"}, "timeout": 1})
|
||||
if !isErr || !strings.Contains(text, "no event families") {
|
||||
t.Errorf("a runtime issued no families: %s", text)
|
||||
@@ -194,7 +194,7 @@ func TestMeshWaitIsAnnouncedAmongSeven(t *testing.T) {
|
||||
}
|
||||
}
|
||||
sort.Strings(names)
|
||||
want := []string{"mesh_call", "mesh_describe", "mesh_machine", "mesh_overview", "mesh_runtimes", "mesh_search", "mesh_wait"}
|
||||
want := []string{"mesh_call", "mesh_describe", "mesh_first_view_again", "mesh_first_views", "mesh_machine", "mesh_overview", "mesh_runtimes", "mesh_search", "mesh_wait"}
|
||||
if !reflect.DeepEqual(names, want) {
|
||||
t.Errorf("announced %v, want %v", names, want)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,195 @@
|
||||
// Package firstview is the tool runner's picture of the four states it reads of the indexer and the assigner: the assigner's
|
||||
// placement and seat-holders, the indexer's manifests and modules (novox/hq ADR 0615 §1, §3; ADR 0621 §2; issue 1966 TLS-5).
|
||||
//
|
||||
// **It never reads the owners' state.** At start it asks each owner's first-view tool until answered (the owner answers in a
|
||||
// runtime of its own, never through this one: ADR 0621 §1), then follows the owner's views family, one event per changed key;
|
||||
// a gap in the revisions, or an owner whose state was made again, asks the first view once more, said in the log. Past two
|
||||
// minutes without an answer it writes a finding in its own state `findings`, which the condition keeper raises and clears.
|
||||
package firstview
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/sdk/views"
|
||||
|
||||
"github.com/novox/mesh-tools/node-tools/internal/bus"
|
||||
)
|
||||
|
||||
// The owners and their first-view tools (the tools' addresses are `<owner>.<tool>`).
|
||||
const (
|
||||
Indexer = "mesh-delivery-indexer"
|
||||
Assigner = "mesh-delivery-assigner"
|
||||
)
|
||||
|
||||
// Spec is one first view the runner holds.
|
||||
type Spec struct{ Owner, Tool, State string }
|
||||
|
||||
// Specs are the four.
|
||||
var Specs = []Spec{
|
||||
{Assigner, "placement", "placement"},
|
||||
{Assigner, "seat-holders", "seat-holders"},
|
||||
{Indexer, "manifests", "manifests"},
|
||||
{Indexer, "modules", "modules"},
|
||||
}
|
||||
|
||||
// Name is the name a state is read by, `<owner>.<state>`: delivery.PlacementRead, delivery.SeatHoldersRead and the indexer's.
|
||||
func (s Spec) Name() string { return s.Owner + "." + s.State }
|
||||
|
||||
// Views is the four followers.
|
||||
type Views struct {
|
||||
logf func(string, ...any)
|
||||
followers map[string]*views.Follower // by Spec.Name()
|
||||
byState map[string]*views.Follower // by the state an event names
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
wg sync.WaitGroup
|
||||
deps Deps
|
||||
}
|
||||
|
||||
// FindingsState is the runner's own state the findings are written in.
|
||||
const FindingsState = "findings"
|
||||
|
||||
// Deps is what the views reach the bus through: Start makes them from the runner's connection, a test from fakes.
|
||||
type Deps struct {
|
||||
// Module is the runner's own name, the observer of its findings.
|
||||
Module string
|
||||
// Ask calls an owner's tool, `<owner>.<tool>`.
|
||||
Ask func(address string, body any) (json.RawMessage, error)
|
||||
// Consume hands each event of the owners' views families to deliver, until the stop is called.
|
||||
Consume func(deliver func(bus.Envelope) error) (stop func(), err error)
|
||||
// PutFinding and DeleteFinding keep and remove a finding in the runner's own state.
|
||||
PutFinding func(key string, value json.RawMessage) error
|
||||
DeleteFinding func(key string) error
|
||||
// Bound and Wait are the follower's (zero: two minutes, five seconds).
|
||||
Bound, Wait time.Duration
|
||||
}
|
||||
|
||||
// Start asks the four first views, each until answered, and follows the owners' views families on the runner's own
|
||||
// consumers. It returns at once: the picture fills as the owners answer.
|
||||
func Start(parent context.Context, conn *bus.Conn, logf func(string, ...any)) (*Views, func()) {
|
||||
me := conn.Module()
|
||||
return StartWith(parent, Deps{Module: me,
|
||||
Ask: func(address string, body any) (json.RawMessage, error) {
|
||||
got, err := conn.Ask(address, body, "")
|
||||
return got.Result, err
|
||||
},
|
||||
Consume: func(deliver func(bus.Envelope) error) (func(), error) { return conn.ConsumeAs(me, deliver) },
|
||||
PutFinding: func(key string, value json.RawMessage) error {
|
||||
_, err := conn.StatePut(me, FindingsState, key, value)
|
||||
return err
|
||||
},
|
||||
DeleteFinding: func(key string) error { return conn.StateDelete(me, FindingsState, key) },
|
||||
}, logf)
|
||||
}
|
||||
|
||||
// StartWith is Start over deps.
|
||||
func StartWith(parent context.Context, d Deps, logf func(string, ...any)) (*Views, func()) {
|
||||
ctx, cancel := context.WithCancel(parent)
|
||||
v := &Views{logf: logf, ctx: ctx, cancel: cancel, followers: map[string]*views.Follower{}, byState: map[string]*views.Follower{}, deps: d}
|
||||
for _, s := range Specs {
|
||||
s := s
|
||||
call := func(tool string, body any) (json.RawMessage, error) { return d.Ask(s.Owner+"."+tool, body) }
|
||||
f := &views.Follower{Owner: s.Owner, Tool: s.Tool, State: s.State, Ask: views.AskVia(call, s.Tool), Logf: logf,
|
||||
Bound: d.Bound, Wait: d.Wait,
|
||||
Raise: func(st views.Status) { v.finding(st, true) },
|
||||
Clear: func(st views.Status) { v.finding(st, false) }}
|
||||
v.followers[s.Name()] = f
|
||||
v.byState[s.State] = f
|
||||
v.wg.Add(1)
|
||||
go func() { defer v.wg.Done(); _ = f.Start(ctx) }()
|
||||
}
|
||||
stop, err := d.Consume(v.hear)
|
||||
if err != nil {
|
||||
logf("[mesh-tools] the owners' views are not followed (%v): the pictures are the first views alone until the runner starts again", err)
|
||||
stop = func() {}
|
||||
}
|
||||
return v, func() { stop(); cancel(); v.wg.Wait() }
|
||||
}
|
||||
|
||||
// hear applies one views event; a gap asks the first view once more.
|
||||
func (v *Views) hear(env bus.Envelope) error {
|
||||
ev, err := views.DecodeEvent(env.Body)
|
||||
if err != nil {
|
||||
return nil // not a views event: nothing to hold, nothing to retry
|
||||
}
|
||||
f := v.byState[ev.State]
|
||||
if f == nil || f.Owner != ev.Owner {
|
||||
return nil
|
||||
}
|
||||
if gap, why := f.Apply(ev); gap {
|
||||
v.logf("[mesh-tools] a gap in %s's %s (%s): its first view %s is asked once more", f.Owner, f.State, why, f.Tool)
|
||||
go func() { _ = f.Again(v.ctx, "a gap: "+why) }()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// finding writes or removes the runner's finding of a first view unanswered past its bound, in its own state.
|
||||
func (v *Views) finding(st views.Status, raised bool) {
|
||||
key, value, err := views.UnansweredFinding(v.deps.Module, st, time.Now())
|
||||
if err != nil {
|
||||
v.logf("[mesh-tools] the finding of the first view %s is not made: %v", st.Tool, err)
|
||||
return
|
||||
}
|
||||
if raised {
|
||||
err = v.deps.PutFinding(key, value)
|
||||
} else {
|
||||
err = v.deps.DeleteFinding(key)
|
||||
}
|
||||
if err != nil {
|
||||
v.logf("[mesh-tools] the finding of the first view %s is not kept: %v", st.Tool, err)
|
||||
}
|
||||
}
|
||||
|
||||
// State is the picture of the state spec names (`<owner>.<state>`), every current value by key; an error says the owner has
|
||||
// not answered yet, so nothing is claimed of it.
|
||||
func (v *Views) State(name string) (map[string][]byte, error) {
|
||||
f := v.followers[name]
|
||||
if f == nil {
|
||||
return nil, fmt.Errorf("the tool runner holds no first view of %s", name)
|
||||
}
|
||||
select {
|
||||
case <-f.Ready():
|
||||
default:
|
||||
return nil, fmt.Errorf("%s's first view %s has not been answered yet", f.Owner, f.Tool)
|
||||
}
|
||||
out := map[string][]byte{}
|
||||
for k, raw := range f.Values() {
|
||||
out[k] = raw
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// Entry is one value of the picture, nil when the owner holds none.
|
||||
func (v *Views) Entry(name, key string) ([]byte, error) {
|
||||
all, err := v.State(name)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return all[key], nil
|
||||
}
|
||||
|
||||
// Status is each first view as it stands, by `<owner>.<state>`, sorted.
|
||||
func (v *Views) Status() []views.Status {
|
||||
var out []views.Status
|
||||
for _, f := range v.followers {
|
||||
out = append(out, f.Status())
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].Owner+out[i].State < out[j].Owner+out[j].State })
|
||||
return out
|
||||
}
|
||||
|
||||
// Again asks every first view once more, by hand, for why.
|
||||
func (v *Views) Again(ctx context.Context, why string) []error {
|
||||
var errs []error
|
||||
for _, f := range v.followers {
|
||||
if err := f.Again(ctx, why); err != nil {
|
||||
errs = append(errs, err)
|
||||
}
|
||||
}
|
||||
return errs
|
||||
}
|
||||
@@ -0,0 +1,168 @@
|
||||
package firstview
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/sdk/views"
|
||||
|
||||
"github.com/novox/mesh-tools/node-tools/internal/bus"
|
||||
)
|
||||
|
||||
// owners is the four owners' first-view tools as a test answers them: a state's values at a revision, or nothing while down.
|
||||
type owners struct {
|
||||
mu sync.Mutex
|
||||
down bool
|
||||
state map[string]map[string]string // tool → key → value
|
||||
rev map[string]uint64
|
||||
asked map[string]int
|
||||
deliver func(bus.Envelope) error
|
||||
finds map[string]bool
|
||||
}
|
||||
|
||||
func newOwners() *owners {
|
||||
return &owners{state: map[string]map[string]string{"placement": {"anchor": `{"p":1}`}, "seat-holders": {"seat": `{"h":1}`},
|
||||
"manifests": {"d1": `{"m":1}`}, "modules": {"web": `{"w":1}`}},
|
||||
rev: map[string]uint64{"placement": 4, "seat-holders": 2, "manifests": 7, "modules": 3}, asked: map[string]int{}, finds: map[string]bool{}}
|
||||
}
|
||||
|
||||
func (o *owners) ask(address string, _ any) (json.RawMessage, error) {
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
tool := address[strings.LastIndex(address, ".")+1:]
|
||||
o.asked[tool]++
|
||||
if o.down {
|
||||
return nil, errors.New("503 no responders")
|
||||
}
|
||||
vals := map[string]json.RawMessage{}
|
||||
for k, v := range o.state[tool] {
|
||||
vals[k] = json.RawMessage(v)
|
||||
}
|
||||
return json.Marshal(views.Answer{V: 1, Owner: "o", State: tool, Revision: o.rev[tool], Created: time.Unix(1000, 0).UTC(), Values: vals})
|
||||
}
|
||||
|
||||
func (o *owners) say(t *testing.T, state, key, value string, revision uint64, gap bool) {
|
||||
t.Helper()
|
||||
ev := views.Event{V: 1, Owner: owner(state), State: state, Key: key, Revision: revision, Created: time.Unix(1000, 0).UTC(), Gap: gap}
|
||||
if value != "" {
|
||||
ev.Value = json.RawMessage(value)
|
||||
}
|
||||
raw, _ := json.Marshal(ev)
|
||||
if err := o.deliver(bus.Envelope{Key: ev.Owner + ".views." + state, Body: raw}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func owner(state string) string {
|
||||
for _, s := range Specs {
|
||||
if s.State == state {
|
||||
return s.Owner
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func eventually(t *testing.T, what string, ok func() bool) {
|
||||
t.Helper()
|
||||
for end := time.Now().Add(3 * time.Second); !ok(); time.Sleep(5 * time.Millisecond) {
|
||||
if time.Now().After(end) {
|
||||
t.Fatal(what)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// issue 1966 TLS-5 (ADR 0615 §2, §4, §5, §6): the runner holds the four states from the owners' first views and their
|
||||
// events alone; nothing is claimed of a state not yet answered; a gap asks the first view once more; an owner that stays down
|
||||
// past the bound is a finding, cleared by the answer; the status and the hand ask show them.
|
||||
func TestTheRunnerHoldsTheFourStatesFromTheOwnersFirstViewsAndTheirEvents(t *testing.T) {
|
||||
o := newOwners()
|
||||
o.down = true
|
||||
var logs []string
|
||||
var lmu sync.Mutex
|
||||
deps := Deps{Module: "node-tools", Ask: o.ask, Bound: 30 * time.Millisecond, Wait: 10 * time.Millisecond,
|
||||
Consume: func(deliver func(bus.Envelope) error) (func(), error) { o.deliver = deliver; return func() {}, nil },
|
||||
PutFinding: func(key string, _ json.RawMessage) error {
|
||||
o.mu.Lock()
|
||||
o.finds[key] = true
|
||||
o.mu.Unlock()
|
||||
return nil
|
||||
},
|
||||
DeleteFinding: func(key string) error {
|
||||
o.mu.Lock()
|
||||
delete(o.finds, key)
|
||||
o.mu.Unlock()
|
||||
return nil
|
||||
}}
|
||||
v, stop := StartWith(context.Background(), deps, func(f string, a ...any) { lmu.Lock(); logs = append(logs, f); lmu.Unlock() })
|
||||
defer stop()
|
||||
|
||||
// Down: nothing is claimed, and past the bound each unanswered view is a finding.
|
||||
if _, err := v.State("mesh-delivery-assigner.placement"); err == nil || !strings.Contains(err.Error(), "not been answered") {
|
||||
t.Fatalf("a state its owner has not answered is claimed: %v", err)
|
||||
}
|
||||
eventually(t, "no finding past the bound", func() bool { o.mu.Lock(); defer o.mu.Unlock(); return len(o.finds) == 4 })
|
||||
|
||||
// Up: answered, the findings are cleared, and the pictures hold the owners' values.
|
||||
o.mu.Lock()
|
||||
o.down = false
|
||||
o.mu.Unlock()
|
||||
eventually(t, "not answered", func() bool { _, err := v.State("mesh-delivery-indexer.manifests"); return err == nil })
|
||||
eventually(t, "findings not cleared", func() bool { o.mu.Lock(); defer o.mu.Unlock(); return len(o.finds) == 0 })
|
||||
if got, _ := v.Entry("mesh-delivery-indexer.modules", "web"); string(got) != `{"w":1}` {
|
||||
t.Errorf("the modules picture holds %q", got)
|
||||
}
|
||||
if got, _ := v.Entry("mesh-delivery-indexer.modules", "none"); got != nil {
|
||||
t.Errorf("an absent key is %q", got)
|
||||
}
|
||||
|
||||
// Events: the next revision is applied; an old one passed over; a deletion removes; none of it asks again.
|
||||
asked := func(tool string) int { o.mu.Lock(); defer o.mu.Unlock(); return o.asked[tool] }
|
||||
before := asked("placement")
|
||||
o.say(t, "placement", "laptop", `{"p":2}`, 5, false)
|
||||
o.say(t, "placement", "laptop", `{"p":0}`, 5, false)
|
||||
o.say(t, "placement", "anchor", "", 6, false)
|
||||
// (a deleted mark carries no value: Deleted is the encoding's, set below)
|
||||
if got, _ := v.Entry("mesh-delivery-assigner.placement", "laptop"); string(got) != `{"p":2}` {
|
||||
t.Errorf("the next revision is not applied: %q", got)
|
||||
}
|
||||
if n := asked("placement"); n != before {
|
||||
t.Errorf("an event in order asked the first view again (%d asks, was %d)", n, before)
|
||||
}
|
||||
|
||||
// A gap (revision 9 where 7 was next) asks the first view once more, said in the log.
|
||||
o.say(t, "placement", "x", `{"p":9}`, 9, false)
|
||||
eventually(t, "a gap did not ask the first view again", func() bool { return asked("placement") > before })
|
||||
|
||||
// The status says owner, tool, revision and the asks; the hand ask asks every view again.
|
||||
var placement views.Status
|
||||
for _, s := range v.Status() {
|
||||
if s.State == "placement" {
|
||||
placement = s
|
||||
}
|
||||
}
|
||||
if placement.Owner != "mesh-delivery-assigner" || placement.Tool != "placement" || placement.Asks < 2 || placement.Answered.IsZero() {
|
||||
t.Errorf("the status of placement is %+v", placement)
|
||||
}
|
||||
manifests := asked("manifests")
|
||||
if errs := v.Again(context.Background(), "doubted"); len(errs) != 0 {
|
||||
t.Fatalf("the hand ask failed: %v", errs)
|
||||
}
|
||||
if asked("manifests") != manifests+1 {
|
||||
t.Errorf("the hand ask did not ask the manifests again")
|
||||
}
|
||||
for _, st := range v.Status() {
|
||||
if st.State == "manifests" && st.ByHand != 1 {
|
||||
t.Errorf("the hand ask is not counted: %+v", st)
|
||||
}
|
||||
}
|
||||
lmu.Lock()
|
||||
defer lmu.Unlock()
|
||||
if !strings.Contains(strings.Join(logs, "\n"), "a gap in") {
|
||||
t.Errorf("the gap was not said in the log: %v", logs)
|
||||
}
|
||||
}
|
||||
@@ -67,14 +67,17 @@
|
||||
]
|
||||
},
|
||||
"state": [
|
||||
"bound"
|
||||
"bound",
|
||||
"findings"
|
||||
],
|
||||
"consumes": [
|
||||
"mesh-delivery-assigner.views.placement",
|
||||
"mesh-delivery-assigner.views.seat-holders",
|
||||
"mesh-delivery-indexer.views.manifests",
|
||||
"mesh-delivery-indexer.views.modules"
|
||||
],
|
||||
"reads": [
|
||||
"mesh-provisioner.provisions",
|
||||
"mesh-provisioner.plans",
|
||||
"mesh-delivery-assigner.seat-holders",
|
||||
"mesh-delivery-assigner.placement",
|
||||
"mesh-delivery-indexer.modules",
|
||||
"mesh-delivery-indexer.manifests"
|
||||
"mesh-provisioner.plans"
|
||||
]
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user