diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go index 746dc902..4f681d8d 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go @@ -18,11 +18,88 @@ import ( "git.novox.be/novox/mesh-sdk/go/provisions" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/busretry" + + "git.novox.be/novox/mesh-delivery/sdk/enginereads" ) +// ManifestReadsTool lists, per node, what its engine states it reads, by asking the node-engine seat's verb of every node +// at once (ADR 0618 §3). +const ManifestReadsTool = "node_manifest_reads" + +// ManifestReadsBound bounds each node's ask of the survey: the whole call stays within a tool call's 30 s. +var ManifestReadsBound = 10 * time.Second + +// askEngine is the survey's caller: the node-engine seat's verb of one machine, through the runtime (stdio.Ask). An error +// that is no lack of an answer is the engine's own words: it answered and does not serve the verb. +func askEngine(node, key string, body any) (json.RawMessage, error) { + raw, err := stdio.Ask(key+"@"+node, body) + if err != nil { + words := strings.ToLower(err.Error()) + if !strings.Contains(words, "no responders") && !strings.Contains(words, "timeout") && !strings.Contains(words, "deadline") { + return nil, &busretry.Answered{Words: err.Error()} + } + } + return raw, err +} + +// engineCaller is how node_manifest_reads asks; a test's fake in its place. +var engineCaller enginereads.Caller = askEngine + +// manifestReads is node_manifest_reads over r's haves and presence: every node with a record or a presence, asked through +// call. +func manifestReads(r view.Reader, call enginereads.Caller) (any, error) { + set := map[string]bool{} + records := map[string][]string{} + if r.HavesState.Keys != nil { + keys, err := r.HavesState.Keys() + if err != nil { + return nil, fmt.Errorf("the haves cannot be read: %w", err) + } + for _, key := range keys { + node, _, isNode, err := delivery.ParseHaveKey(key) + if err != nil || !isNode { + continue + } + set[node] = true + if raw, err := r.HavesState.Get(key); err == nil && raw != nil { + if h, err := delivery.DecodeHaveNode(raw); err == nil && h.Record != nil { + records[node] = h.Record.ManifestReads + } + } + } + } + if r.PresenceState.Keys != nil { + if keys, err := r.PresenceState.Keys(); err == nil { + for _, key := range keys { + if node, part, _, err := delivery.ParsePresenceKey(key); err == nil && part == "" && node != "" { + set[node] = true + } + } + } + } + nodes := make([]string, 0, len(set)) + for node := range set { + nodes = append(nodes, node) + } + rows := enginereads.Survey(call, nodes, func(node string) ([]string, bool) { + reads, ok := records[node] + return reads, ok + }, ManifestReadsBound, time.Now()) + return map[string]any{"nodes": rows}, nil +} + // Tools are the read tools over a modules state, and build_usage over the build modules' states. func Tools(r view.Reader, u view.UsageReader) []stdio.Tool { return []stdio.Tool{ + {Name: ManifestReadsTool, + Description: "What each node's engine states it reads of a manifest (manifest words, forms and seat names, novox/hq ADR 0618 " + + "§3), for the whole mesh in one call: every node with a record or a presence is asked the node-engine seat's verb " + + "manifest-reads at once, each within " + ManifestReadsBound.String() + ". Per node: stated (with the capabilities and " + + "when the engine stated them), not served (the engine is built before the verb) or away (nobody answered), and for the " + + "last two the reads its node record states, which the assigner reads in their place. Read-only.", + Input: map[string]any{"type": "object", "properties": map[string]any{}}, + Run: func(map[string]any) (any, error) { return manifestReads(r, engineCaller) }}, {Name: view.StatusTool, Description: "The mesh's status, derived by the view from the buckets it reads (novox/hq ADR 0564 §10), nothing written as a " + "summary: every open condition of the condition keeper (mesh-conditions), urgent first then oldest; what is quiet (each " + diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go index 144d9d30..01a8f924 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go @@ -11,8 +11,11 @@ import ( "time" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/busretry" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-delivery/sdk/enginereads" + "mesh-delivery-view/internal/view" ) @@ -56,6 +59,7 @@ const ( func TestModuleToolAnswersTheStoredValue(t *testing.T) { t.Run("indexer_packages answers the published modules", packagesToolAnswersThePublishedModules) t.Run("view_tiers names each module's wait", tiersToolSaysEachModulesWait) + t.Run("node_manifest_reads asks every node's engine in one call", manifestReadsAsksEveryNode) r := fake{"a": withFP2, "b": withNone}.reader() for key, want := range map[string]string{"a": withFP2, "b": withNone} { out, err := run(t, r, view.ModuleReadTool, map[string]any{"module": key}) @@ -570,3 +574,48 @@ func packagesToolAnswersThePublishedModules(t *testing.T) { t.Error("a path that is no mesh Go module's was read") } } + +// manifestReadsAsksEveryNode: node_manifest_reads lists each node's capabilities by asking the node-engine seat's verb +// (ADR 0618 §3) of every node with a record or a presence, at once, each within its bound: stated, not served (with its +// record's reads, which the assigner reads in their place) or away (a node that never answers holds no other). +func manifestReadsAsksEveryNode(t *testing.T) { + was, bound := engineCaller, ManifestReadsBound + defer func() { engineCaller, ManifestReadsBound = was, bound }() + ManifestReadsBound = 200 * time.Millisecond + hang := make(chan struct{}) + defer close(hang) + engineCaller = func(node, key string, _ any) (json.RawMessage, error) { + switch { + case key != "seat:node-engine.manifest-reads": + return nil, fmt.Errorf("asked %s", key) + case node == "g14": + return nil, &busretry.Answered{Words: "the seat node-engine serves no verb manifest-reads"} + case node == "novox": + <-hang // never answers within the bound + return nil, nil + } + return delivery.EncodeManifestReadsAnswer(delivery.ManifestReadsAnswer{Node: node, ManifestReads: []string{"form:new"}, StatedAt: time.Now()}) + } + record := func(node string, reads ...string) string { + raw, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: node, ManifestReads: reads}}) + return string(raw) + } + on, _ := delivery.EncodePresence(delivery.Presence{Online: true}) + r := view.Reader{HavesState: fake{"ace.node": record("ace"), "g14.node": record("g14", "word:module"), "ace.web": "{}"}.reader().State, + PresenceState: fake{"novox": string(on)}.reader().State} + started := time.Now() + out, err := run(t, r, ManifestReadsTool, nil) + if err != nil || time.Since(started) > 2*time.Second { + t.Fatalf("node_manifest_reads: %v after %s", err, time.Since(started)) + } + rows, _ := out.(map[string]any)["nodes"].([]enginereads.SurveyRow) + byNode := map[string]enginereads.SurveyRow{} + for _, row := range rows { + byNode[row.Node] = row + } + if len(rows) != 3 || byNode["ace"].Answer != enginereads.Stated || byNode["ace"].ManifestReads[0] != "form:new" || + byNode["g14"].Answer != enginereads.NotServed || !byNode["g14"].RecordRead || byNode["g14"].Record[0] != "word:module" || + byNode["novox"].Answer != enginereads.Away || byNode["novox"].RecordRead { + t.Errorf("node_manifest_reads = %+v", rows) + } +} diff --git a/modules/mesh-delivery-view/go.mod b/modules/mesh-delivery-view/go.mod index 035f39f7..82d12a43 100644 --- a/modules/mesh-delivery-view/go.mod +++ b/modules/mesh-delivery-view/go.mod @@ -24,3 +24,5 @@ require ( require git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000 replace git.novox.be/novox/mesh-delivery/sdk => ../../sdk + +replace git.novox.be/novox/mesh-sdk/go => ../../../0618-sdk/go diff --git a/modules/mesh-delivery-view/go.sum b/modules/mesh-delivery-view/go.sum index 7679a5dc..98dc72c4 100644 --- a/modules/mesh-delivery-view/go.sum +++ b/modules/mesh-delivery-view/go.sum @@ -1,5 +1,3 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8 h1:XEbIVImp6yGjA91rp0iZBoDdL5KfamjOuDLwWvqh0hc= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8/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= diff --git a/modules/mesh-delivery-view/manifest_test.go b/modules/mesh-delivery-view/manifest_test.go index 4e65b88f..ee78cba2 100644 --- a/modules/mesh-delivery-view/manifest_test.go +++ b/modules/mesh-delivery-view/manifest_test.go @@ -53,6 +53,7 @@ type manifest struct { Reads []string `json:"reads"` Uses []string `json:"uses"` Tools []string `json:"tools"` + Invokes []string `json:"invokes"` Emits []string `json:"emits"` OwnSecrets map[string]string `json:"own-secrets"` SecretsOwner string `json:"secrets-owner"` @@ -119,9 +120,13 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 } // The two read tools of the indexer's modules (issue 1562), carried by a bundle the runtime launches; the indexer // itself serves no tool (ADR 0437 §4). Carrying them gives the tool runner read access to every bucket the view reads. - if want := []string{"view_status", "indexer_module", "indexer_modules", "indexer_candidate", "indexer_candidates", "indexer_predefined", "indexer_config_candidates", "indexer_packages", "builder_build", "builder_builds", "builder_history", "assigner_wants", "assigner_want", "assigner_differences", "node_haves", "node_trusts", "build_usage", "view_reach", "indexer_module_history", "indexer_heard", "view_tiers"}; !reflect.DeepEqual(m.Tools, want) { + if want := []string{"view_status", "indexer_module", "indexer_modules", "indexer_candidate", "indexer_candidates", "indexer_predefined", "indexer_config_candidates", "indexer_packages", "builder_build", "builder_builds", "builder_history", "assigner_wants", "assigner_want", "assigner_differences", "node_haves", "node_trusts", "build_usage", "view_reach", "indexer_module_history", "indexer_heard", "view_tiers", "node_manifest_reads"}; !reflect.DeepEqual(m.Tools, want) { t.Errorf("tools = %v, want %v", m.Tools, want) } + // ADR 0618 §3: node_manifest_reads asks the node-engine seat's verb of every node, and the view invokes nothing else. + if want := []string{"seat:node-engine.manifest-reads"}; !reflect.DeepEqual(m.Invokes, want) { + t.Errorf("invokes = %v, want %v", m.Invokes, want) + } for _, a := range m.Build.Artifacts { // A bundle compiled to a binary is one executable: what the runtime loads is that binary's name, or the controller's // module check refuses the manifest (mesh-sdk catalogue build.go, ADR 0193). diff --git a/modules/mesh-delivery-view/module.json b/modules/mesh-delivery-view/module.json index 55bbbfb6..14523359 100644 --- a/modules/mesh-delivery-view/module.json +++ b/modules/mesh-delivery-view/module.json @@ -24,7 +24,11 @@ "view_reach", "indexer_module_history", "indexer_heard", - "view_tiers" + "view_tiers", + "node_manifest_reads" + ], + "invokes": [ + "seat:node-engine.manifest-reads" ], "emits": [ "delivered", diff --git a/sdk/enginereads/enginereads.go b/sdk/enginereads/enginereads.go index 110f5bb8..98aac884 100644 --- a/sdk/enginereads/enginereads.go +++ b/sdk/enginereads/enginereads.go @@ -493,3 +493,66 @@ func Tool(name, owner string, st State, drops bool, now func() time.Time) stdio. return Read(st, node) }} } + +// How a node answered a survey (Survey). +const ( + // Stated: the engine answered the verb. + Stated = "stated" + // NotServed: the engine answered, but does not serve the verb (an engine built before it). + NotServed = "not served" + // Away: nobody answered within the bound: the node is away, or its engine does not serve the verb and nothing else + // answers for it. + Away = "away" +) + +// SurveyRow is one node's row of a survey: how it answered, its statement when it stated one, and its record's reads +// where it did not (the fallback the assigner reads in its place). +type SurveyRow struct { + Node string `json:"node"` + Answer string `json:"answer"` + ManifestReads []string `json:"manifest-reads,omitempty"` + StatedAt time.Time `json:"stated-at,omitzero"` + Record []string `json:"record,omitempty"` + RecordRead bool `json:"record-read,omitempty"` + Words string `json:"words,omitempty"` +} + +// Survey asks every node's engine the verb at once, each ask bounded by bound, so the whole survey takes at most about +// bound however many nodes there are; a node that did not answer is named with its record's reads where record reads +// one. Sorted by node. +func Survey(call Caller, nodes []string, record func(node string) ([]string, bool), bound time.Duration, now time.Time) []SurveyRow { + rows := make([]SurveyRow, len(nodes)) + var wg sync.WaitGroup + for i, node := range nodes { + wg.Add(1) + go func() { + defer wg.Done() + got := make(chan Statement, 1) + go func() { got <- Ask(call, node, now) }() + var s Statement + wait, cancel := context.WithTimeout(context.Background(), bound) + defer cancel() + select { + case s = <-got: + case <-wait.Done(): + s = Statement{Node: node, Words: fmt.Sprintf("no answer within %s", bound)} + } + row := SurveyRow{Node: node, Words: s.Words} + switch { + case s.From == FromVerb: + row.Answer, row.ManifestReads, row.StatedAt = Stated, s.ManifestReads, s.StatedAt + case s.Answered: + row.Answer = NotServed + default: + row.Answer = Away + } + if row.Answer != Stated && record != nil { + row.Record, row.RecordRead = record(node) + } + rows[i] = row + }() + } + wg.Wait() + sort.Slice(rows, func(i, j int) bool { return rows[i].Node < rows[j].Node }) + return rows +}