novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

mesh-delivery-view: node_manifest_reads lists what each node's engine reads by asking the node-engine seat's verb of every node at once, each bounded, naming a node not served or away with its record's reads (ADR 0618 §3, issue 774)

Browse Source 
Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn
This commit is contained in:
jschoubben
2026-11-04 12:26:48 +01:00
parent ad77815d88
commit e41666d846
7 changed files with 202 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go
+77
View File
@@ -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 " +
modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go
+49
View File
@@ -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)
}
}
modules/mesh-delivery-view/go.mod
+2
View File
@@ -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
modules/mesh-delivery-view/go.sum
-2
View File
@@ -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=
modules/mesh-delivery-view/manifest_test.go
+6 -1
View File
@@ -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).
modules/mesh-delivery-view/module.json
+5 -1
View File
@@ -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",
sdk/enginereads/enginereads.go
+63
View File
@@ -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
}