diff --git a/node-tools/cmd/node-tools/main.go b/node-tools/cmd/node-tools/main.go index bf55106..7c821fc 100644 --- a/node-tools/cmd/node-tools/main.go +++ b/node-tools/cmd/node-tools/main.go @@ -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) } diff --git a/node-tools/go.mod b/node-tools/go.mod index d88a8a1..94262f0 100644 --- a/node-tools/go.mod +++ b/node-tools/go.mod @@ -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 diff --git a/node-tools/internal/console/address.go b/node-tools/internal/console/address.go index e3f8aaf..f4604b5 100644 --- a/node-tools/internal/console/address.go +++ b/node-tools/internal/console/address.go @@ -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 `.` (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) } diff --git a/node-tools/internal/console/address_test.go b/node-tools/internal/console/address_test.go index fec01f4..627de7a 100644 --- a/node-tools/internal/console/address_test.go +++ b/node-tools/internal/console/address_test.go @@ -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) } diff --git a/node-tools/internal/console/client.go b/node-tools/internal/console/client.go index 751731b..dc1ef15 100644 --- a/node-tools/internal/console/client.go +++ b/node-tools/internal/console/client.go @@ -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 } diff --git a/node-tools/internal/console/console_test.go b/node-tools/internal/console/console_test.go index 2ad8e70..4a61f34 100644 --- a/node-tools/internal/console/console_test.go +++ b/node-tools/internal/console/console_test.go @@ -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) } } diff --git a/node-tools/internal/console/empty_round_test.go b/node-tools/internal/console/empty_round_test.go index 474dafa..bff7b5b 100644 --- a/node-tools/internal/console/empty_round_test.go +++ b/node-tools/internal/console/empty_round_test.go @@ -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) diff --git a/node-tools/internal/console/firstviews.go b/node-tools/internal/console/firstviews.go new file mode 100644 index 0000000..8f9930d --- /dev/null +++ b/node-tools/internal/console/firstviews.go @@ -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) +} diff --git a/node-tools/internal/console/firstviews_test.go b/node-tools/internal/console/firstviews_test.go new file mode 100644 index 0000000..3a8e2da --- /dev/null +++ b/node-tools/internal/console/firstviews_test.go @@ -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)) + } + } +} diff --git a/node-tools/internal/console/manifest_waits_test.go b/node-tools/internal/console/manifest_waits_test.go index 4d6d2e4..d651d68 100644 --- a/node-tools/internal/console/manifest_waits_test.go +++ b/node-tools/internal/console/manifest_waits_test.go @@ -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) } } diff --git a/node-tools/internal/console/mcp.go b/node-tools/internal/console/mcp.go index 66a196a..971c38f 100644 --- a/node-tools/internal/console/mcp.go +++ b/node-tools/internal/console/mcp.go @@ -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 } diff --git a/node-tools/internal/console/runtimes_test.go b/node-tools/internal/console/runtimes_test.go index 5287279..94af57d 100644 --- a/node-tools/internal/console/runtimes_test.go +++ b/node-tools/internal/console/runtimes_test.go @@ -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) } diff --git a/node-tools/internal/console/source_test.go b/node-tools/internal/console/source_test.go new file mode 100644 index 0000000..a3fed70 --- /dev/null +++ b/node-tools/internal/console/source_test.go @@ -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 +} diff --git a/node-tools/internal/console/wait_test.go b/node-tools/internal/console/wait_test.go index 40c5a0c..862517f 100644 --- a/node-tools/internal/console/wait_test.go +++ b/node-tools/internal/console/wait_test.go @@ -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) } diff --git a/node-tools/internal/firstview/firstview.go b/node-tools/internal/firstview/firstview.go new file mode 100644 index 0000000..cf6db43 --- /dev/null +++ b/node-tools/internal/firstview/firstview.go @@ -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 `.`). +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, `.`: 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, `.`. + 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 (`.`), 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 `.`, 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 +} diff --git a/node-tools/internal/firstview/firstview_test.go b/node-tools/internal/firstview/firstview_test.go new file mode 100644 index 0000000..0fddaa4 --- /dev/null +++ b/node-tools/internal/firstview/firstview_test.go @@ -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) + } +} diff --git a/node-tools/module.json b/node-tools/module.json index 10f153a..f3bec56 100644 --- a/node-tools/module.json +++ b/node-tools/module.json @@ -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" ] }