diff --git a/go/clients/nodeengineclient/enginereads/enginereads.go b/go/clients/nodeengineclient/enginereads/enginereads.go new file mode 100644 index 0000000..4ae20b2 --- /dev/null +++ b/go/clients/nodeengineclient/enginereads/enginereads.go @@ -0,0 +1,643 @@ +// Package enginereads is how a mesh-delivery part learns what each node's engine reads of a manifest (novox/hq ADR 0618 +// §3, §4): it asks the node seat node-engine's verb manifest-reads, on the part's start, on the seat's event +// manifest-reads-changed for that node and when a person asks again, and keeps each node's last answer in a state of its +// own, one key per node, so a node that is away keeps the statement it last made (ADR 0624 §1). The assigner, the +// builder and the delivery view share it; it is the seat node-engine's client's, not mesh-sdk's (ADR 0634 §1, §2). +package enginereads + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "slices" + "sort" + "strings" + "sync" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/busretry" + + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" +) + +// The verb and the event as the seat's contract names them (mesh-contracts seats/nodeengine), as a manifest names them, +// and the subject the event is said on with the engine's machine last. +const ( + Verb = nodeengine.VerbManifestReads + Event = nodeengine.EventManifestReadsChanged + Invoke = "seat:" + nodeengine.Seat + "." + Verb + Consumes = nodeengine.Seat + "." + Event +) + +// SubjectPrefix is the event's subject up to the machine, its last token. +var SubjectPrefix = strings.TrimSuffix(nodeengine.SubjectManifestReadsChanged("*"), "*") + +// The reasons an engine says its reads, in the event's why (ADR 0618 §4). +const ( + // ReadsStarted: the engine started, or what it states changed. + ReadsStarted = "started" + // ReadsReturned: the engine's node returned after being away. + ReadsReturned = "returned" +) + +// ContractVersion is the layout version of the verb's answer and the event's body this reader reads. +const ContractVersion = 1 + +// checkNode refuses a node that is no subject token. +func checkNode(node string) error { + if node == "" || strings.ContainsAny(node, ".*> \t\n") { + return fmt.Errorf("%q is no node", node) + } + return nil +} + +// checkReads refuses an empty reader capability. +func checkReads(node string, reads []string) error { + for _, c := range reads { + if strings.TrimSpace(c) == "" { + return fmt.Errorf("the reads of %s name an empty reader capability", node) + } + } + return nil +} + +// DecodeAnswer reads the verb's answer as the contract lays it out, and judges it: the layout's version, the node a +// name, no empty capability, and its stated-at an RFC 3339 time (the contract carries no checks; its reader does). +func DecodeAnswer(raw []byte) (nodeengine.ManifestReadsAnswer, time.Time, error) { + var a nodeengine.ManifestReadsAnswer + if err := json.Unmarshal(raw, &a); err != nil { + return a, time.Time{}, err + } + if a.V != ContractVersion { + return a, time.Time{}, fmt.Errorf("the answer's layout is version %d, not %d", a.V, ContractVersion) + } + if err := checkNode(a.Node); err != nil { + return a, time.Time{}, err + } + if err := checkReads(a.Node, a.ManifestReads); err != nil { + return a, time.Time{}, err + } + at, err := time.Parse(time.RFC3339Nano, a.StatedAt) + if err != nil { + return a, time.Time{}, fmt.Errorf("stated-at %q is no RFC 3339 time", a.StatedAt) + } + return a, at, nil +} + +// DecodeChanged reads the event's body as the contract lays it out, and judges it as DecodeAnswer does, its why one of +// ReadsStarted and ReadsReturned. +func DecodeChanged(raw []byte) (nodeengine.ManifestReadsChangedEvent, error) { + var e nodeengine.ManifestReadsChangedEvent + if err := json.Unmarshal(raw, &e); err != nil { + return e, err + } + if e.V != ContractVersion { + return e, fmt.Errorf("the event's layout is version %d, not %d", e.V, ContractVersion) + } + if err := checkNode(e.Node); err != nil { + return e, err + } + if e.Why != ReadsStarted && e.Why != ReadsReturned { + return e, fmt.Errorf("the reads of %s changed for %q, not %s or %s", e.Node, e.Why, ReadsStarted, ReadsReturned) + } + if err := checkReads(e.Node, e.ManifestReads); err != nil { + return e, err + } + if _, err := time.Parse(time.RFC3339Nano, e.At); err != nil { + return e, fmt.Errorf("at %q is no RFC 3339 time", e.At) + } + return e, nil +} + +// Where a node's statement came from. +const ( + // FromVerb: the node's engine answered the verb. + FromVerb = "verb" + // FromRecord: the engine did not answer it (an engine built before the verb, or nobody answered) and the node's record + // `.node` was read once in its place (the fallback, until every counted engine serves the verb; issue 774). + FromRecord = "record" + // FromNone: neither said anything, so the node is read as stating the baseline (ADR 0614 §4). + FromNone = "none" +) + +// Version is the version of a statement this package writes and the only one it reads. +const Version = 1 + +// Statement is one node's key: what its engine states it reads, when it stated it, when it was asked, where it came +// from, whether the engine answered that ask (it is present), and, while a person's ask again waits, that ask. +type Statement struct { + V int `json:"v"` + Node string `json:"node"` + ManifestReads []string `json:"manifest-reads,omitempty"` + StatedAt time.Time `json:"stated-at,omitzero"` + AskedAt time.Time `json:"asked-at,omitzero"` + From string `json:"from"` + Answered bool `json:"answered"` + Words string `json:"words,omitempty"` + AskAgain *Again `json:"ask-again,omitempty"` +} + +// Again is a person's ask again of one node, written by the tool and answered by the part that keeps the state. +type Again struct { + By string `json:"by"` + Why string `json:"why"` + At time.Time `json:"at"` +} + +// Encode is s as JSON at Version. +func Encode(s Statement) ([]byte, error) { + if s.Node == "" || strings.ContainsAny(s.Node, ".*> ") { + return nil, fmt.Errorf("enginereads: %q is no node", s.Node) + } + s.V = Version + return json.Marshal(s) +} + +// Decode reads a statement; it refuses another version and a value naming no node. +func Decode(raw []byte) (Statement, error) { + var s Statement + if err := json.Unmarshal(raw, &s); err != nil { + return Statement{}, err + } + if s.V != Version { + return Statement{}, fmt.Errorf("enginereads: version %d, not %d", s.V, Version) + } + if s.Node == "" { + return Statement{}, errors.New("enginereads: a statement names no node") + } + return s, nil +} + +// Caller calls a seat's verb on one node: process.Conn.CallOn. +type Caller func(node, key string, body any) (json.RawMessage, error) + +// Ask asks node's engine the verb once and says what it answered. An engine that answered the call with an error is +// present and states nothing through it; one that did not answer is not present. +func Ask(call Caller, node string, now time.Time) Statement { + s := Statement{Node: node, AskedAt: now.UTC(), From: FromNone} + raw, err := call(node, Invoke, map[string]any{}) + var answered *busretry.Answered + switch { + case err == nil: + a, at, derr := DecodeAnswer(raw) + if derr == nil && a.Node != node { + derr = fmt.Errorf("it names the node %s", a.Node) + } + if derr != nil { + s.Answered, s.Words = true, "the engine answered what is no statement: "+derr.Error() + return s + } + s.ManifestReads, s.StatedAt, s.From, s.Answered = slices.Clone(a.ManifestReads), at.UTC(), FromVerb, true + case errors.As(err, &answered): + s.Answered, s.Words = true, "the engine does not serve "+Verb+": "+answered.Words + default: + s.Words = "nobody answered " + Verb + ": " + err.Error() + } + return s +} + +// NodeOf is the node an event of manifest-reads-changed is about, read from its subject's last token and held equal to +// its body's; ok false for a subject of another event. +func NodeOf(subject string, data []byte) (string, bool, error) { + node, ok := strings.CutPrefix(subject, SubjectPrefix) + if !ok || node == "" || strings.Contains(node, ".") { + return "", false, nil + } + e, err := DecodeChanged(data) + if err != nil { + return node, true, err + } + if e.Node != node { + return node, true, fmt.Errorf("the event on %s names the node %s", subject, e.Node) + } + return node, true, nil +} + +// Keeper keeps every node's statement for one part: in memory, and in the part's state when Put is set, so a statement +// outlives the part's restart. Nothing in it waits on a timer: a node is asked when the part says so. +type Keeper struct { + // Call asks the verb; nil asks nothing (every node then reads as the fallback or the baseline). + Call Caller + // Fallback reads a node's record `.node` once where its engine did not answer the verb; nil reads none. + Fallback func(node string) ([]string, bool) + // Put writes a node's key of the part's state, Delete removes it; nil keeps the statements in memory only. + Put func(node string, value []byte) error + Delete func(node string) error + // Changed is told after a node's statement changed. + Changed func() + Logf func(string, ...any) + Now func() time.Time + // Sync asks in the caller's goroutine (tests); otherwise each ask runs on its own. + Sync bool + + mu sync.Mutex + by map[string]Statement + asking map[string]bool +} + +func (k *Keeper) logf(f string, a ...any) { + if k.Logf != nil { + k.Logf(f, a...) + } +} + +func (k *Keeper) now() time.Time { + if k.Now != nil { + return k.Now() + } + return time.Now() +} + +// AskNode asks node's engine, unless an ask of it runs already; why is said in the log. +func (k *Keeper) AskNode(node, why string) { + k.mu.Lock() + if k.asking == nil { + k.asking = map[string]bool{} + } + if k.asking[node] { + k.mu.Unlock() + return + } + k.asking[node] = true + k.mu.Unlock() + if k.Sync { + k.ask(node, why) + return + } + go k.ask(node, why) +} + +func (k *Keeper) ask(node, why string) { + defer func() { + k.mu.Lock() + delete(k.asking, node) + k.mu.Unlock() + }() + s := Statement{Node: node, AskedAt: k.now().UTC(), From: FromNone, Words: "nothing asks the verb"} + if k.Call != nil { + s = Ask(k.Call, node, k.now()) + } + if s.From != FromVerb && k.Fallback != nil { + if reads, ok := k.Fallback(node); ok { + s.ManifestReads, s.From = slices.Clone(reads), FromRecord + k.logf("%s on %s (%s): %s; its record %s.node is read once in its place (until every counted engine serves the verb)", + Verb, node, why, s.Words, node) + } + } + if s.From == FromNone { + k.logf("%s on %s (%s): %s; the node reads as its last statement, or the baseline", Verb, node, why, s.Words) + } + k.take(s) +} + +// take keeps s, unless the node did not answer and an earlier statement stands: an away node keeps its last one (ADR 0624 +// §1), marked as not answering. +func (k *Keeper) take(s Statement) { + k.mu.Lock() + if k.by == nil { + k.by = map[string]Statement{} + } + if old, ok := k.by[s.Node]; ok && !s.Answered && old.From != FromNone { + old.Answered, old.AskedAt, old.Words, old.AskAgain = false, s.AskedAt, s.Words, nil + s = old + } + k.by[s.Node] = s + k.mu.Unlock() + if k.Put != nil { + raw, err := Encode(s) + if err == nil { + err = k.Put(s.Node, raw) + } + if err != nil { + k.logf("the statement of %s could not be kept: %v", s.Node, err) + } + } + if k.Changed != nil { + k.Changed() + } +} + +// Heard takes a change of the part's own state in: at the start, every node's last statement; later, a person's ask again +// (asked now) or a key a person dropped (forgotten). The part's own writes come back here and change nothing. +func (k *Keeper) Heard(node string, raw []byte, deleted bool) { + if deleted { + k.mu.Lock() + _, had := k.by[node] + delete(k.by, node) + k.mu.Unlock() + if had && k.Changed != nil { + k.Changed() + } + return + } + s, err := Decode(raw) + if err != nil || s.Node != node { + k.logf("the key %s of the engine reads is no statement of it, left: %v", node, err) + return + } + if s.AskAgain != nil { + k.AskNode(node, fmt.Sprintf("asked again by %s: %s", s.AskAgain.By, s.AskAgain.Why)) + return + } + k.mu.Lock() + if k.by == nil { + k.by = map[string]Statement{} + } + _, had := k.by[node] + if !had { + k.by[node] = s + } + k.mu.Unlock() + if !had && k.Changed != nil { + k.Changed() + } +} + +// Forget drops a node the part no longer counts (it left the placement): its key too. +func (k *Keeper) Forget(node string) { + k.mu.Lock() + _, had := k.by[node] + delete(k.by, node) + k.mu.Unlock() + if had && k.Delete != nil { + if err := k.Delete(node); err != nil { + k.logf("the statement of %s could not be removed: %v", node, err) + } + } +} + +// Known says whether the part holds a statement of node, from any source. +func (k *Keeper) Known(node string) bool { + k.mu.Lock() + defer k.mu.Unlock() + _, ok := k.by[node] + return ok +} + +// Statements is every node's statement, by node. +func (k *Keeper) Statements() map[string]Statement { + k.mu.Lock() + defer k.mu.Unlock() + out := make(map[string]Statement, len(k.by)) + for n, s := range k.by { + out[n] = s + } + return out +} + +// Reads is, of the nodes given, what each states it reads: its last statement, nil (the baseline) for a node with none. +func (k *Keeper) Reads(nodes []string) map[string][]string { + k.mu.Lock() + defer k.mu.Unlock() + out := make(map[string][]string, len(nodes)) + for _, n := range nodes { + out[n] = slices.Clone(k.by[n].ManifestReads) + } + return out +} + +// Present is, of the nodes given, those whose engine answered when last asked, with what each states. +func (k *Keeper) Present(nodes []string) map[string][]string { + k.mu.Lock() + defer k.mu.Unlock() + out := map[string][]string{} + for _, n := range nodes { + if s, ok := k.by[n]; ok && s.Answered { + out[n] = slices.Clone(s.ManifestReads) + } + } + return out +} + +// State is what the tool needs of the part's state: the runtime's (stdio.KeptState through KeptState), or a test's. +type State interface { + Keys() ([]string, error) + Get(key string) ([]byte, uint64, bool, error) + Put(key string, value []byte) (uint64, error) + Delete(key string) error + // Await waits at most bound for a revision of key after after whose value done says is the answer. + Await(key string, after uint64, done func([]byte) bool, bound time.Duration) ([]byte, error) +} + +// Read lists every node's statement in st, by node name, or one node's; a key that does not read is named with why. +func Read(st State, node string) (map[string]any, error) { + keys := []string{node} + if node == "" { + var err error + if keys, err = st.Keys(); err != nil { + return nil, err + } + } + sort.Strings(keys) + nodes := map[string]any{} + for _, key := range keys { + raw, _, found, err := st.Get(key) + switch { + case err != nil: + return nil, err + case !found: + nodes[key] = map[string]any{"from": FromNone, "words": "no statement is kept: read as the baseline"} + continue + } + s, derr := Decode(raw) + if derr != nil { + nodes[key] = map[string]any{"unread": derr.Error()} + continue + } + nodes[key] = s + } + return map[string]any{"nodes": nodes}, nil +} + +// AskAgainOf writes node's key with a person's ask again and waits at most bound for the part's answer: the next +// statement it writes there. +func AskAgainOf(st State, node, by, why string, now time.Time, bound time.Duration) (Statement, error) { + if node == "" || strings.ContainsAny(node, ".*> ") { + return Statement{}, fmt.Errorf("%q is no node", node) + } + if strings.TrimSpace(why) == "" { + return Statement{}, errors.New("say why it is asked again") + } + s := Statement{Node: node, From: FromNone} + if raw, _, found, err := st.Get(node); err != nil { + return Statement{}, err + } else if found { + if old, err := Decode(raw); err == nil { + s = old + } + } + s.AskAgain = &Again{By: by, Why: why, At: now.UTC()} + raw, err := Encode(s) + if err != nil { + return Statement{}, err + } + rev, err := st.Put(node, raw) + if err != nil { + return Statement{}, err + } + answer, err := st.Await(node, rev, func(v []byte) bool { + got, err := Decode(v) + return err == nil && got.AskAgain == nil + }, bound) + if err != nil { + return Statement{}, fmt.Errorf("the part did not answer the ask again of %s within %s: %w", node, bound, err) + } + return Decode(answer) +} + +// KeptState is a part's state as the runtime keeps it, for the tool. +func KeptState(s stdio.KeptState) State { return keptState{s} } + +type keptState struct{ s stdio.KeptState } + +func (k keptState) Keys() ([]string, error) { return k.s.Keys() } + +func (k keptState) Get(key string) ([]byte, uint64, bool, error) { + e, err := k.s.Get(key) + if err != nil || e == nil { + return nil, 0, false, err + } + return e.Value, e.Revision, true, nil +} + +func (k keptState) Put(key string, value []byte) (uint64, error) { + return k.s.Put(key, json.RawMessage(value)) +} + +func (k keptState) Delete(key string) error { return k.s.Delete(key) } + +// Await watches key for the answer, and reads it once too in case the watch heard it before it began; a context's bound, +// no timer (ADR 0437 §8). +func (k keptState) Await(key string, after uint64, done func([]byte) bool, bound time.Duration) ([]byte, error) { + heard := make(chan []byte, 1) + if err := k.s.Watch(key, func(c stdio.StateChange) error { + if c.Op == "put" && c.Revision > after && done(c.Value) { + select { + case heard <- c.Value: + default: + } + } + return nil + }); err != nil { + return nil, err + } + if e, err := k.s.Get(key); err == nil && e != nil && e.Revision > after && done(e.Value) { + return e.Value, nil + } + wait, cancel := context.WithTimeout(context.Background(), bound) + defer cancel() + select { + case v := <-heard: + return v, nil + case <-wait.Done(): + return nil, errors.New("it is not running, or cannot read the verb's answer") + } +} + +// AnswerWithin is how long the tool waits for the part to answer an ask again: the verb's own call bound and a margin. +var AnswerWithin = 25 * time.Second + +// Tool is a part's tool over its state of engine reads (tools-first, ADR 0618 §3): read every node's statement or one, +// ask one node's engine again now, and, where drops, drop one node's key (a node the part keeps by mistake). owner names +// the part in the words. +func Tool(name, owner string, st State, drops bool, now func() time.Time) stdio.Tool { + props := map[string]any{ + "node": map[string]any{"type": "string", "description": "one node; every node when left out"}, + "ask_again": map[string]any{"type": "string", "description": "\"true\": ask the node's engine " + Verb + " again now, and answer what it states"}, + "why": map[string]any{"type": "string", "description": "why it is asked again or dropped, in plain words"}, + } + description := "What each node's engine states it reads of a manifest (manifest words, forms and seat names, novox/hq ADR 0618 §3), " + + "as " + owner + " last asked the node-engine seat's verb " + Verb + " and uses it: per node its capabilities, when the engine " + + "stated them and when it was asked, whether it answered (present) and where the statement came from: verb, record (the " + + "engine did not answer and its node record was read once in its place) or none (read as the baseline). A node that is " + + "away keeps its last statement. With node and ask_again, " + owner + " asks that node's engine again and answers its new " + + "statement." + if drops { + props["drop"] = map[string]any{"type": "string", "description": "\"true\": drop the node's key, so it reads as stating the baseline until its engine is asked again"} + description += " With node and drop, the node's key is dropped." + } + return stdio.Tool{Name: name, Description: description, Input: map[string]any{"type": "object", "properties": props}, + Run: func(args map[string]any) (any, error) { + str := func(k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) } + node := str("node") + switch { + case str("ask_again") == "true": + return AskAgainOf(st, node, stdio.CallerOf(args), str("why"), now(), AnswerWithin) + case drops && str("drop") == "true": + if node == "" || str("why") == "" { + return nil, errors.New("drop takes node and why") + } + if err := st.Delete(node); err != nil { + return nil, err + } + return map[string]any{"node": node, "dropped": true, + "next": "the node reads as stating the baseline until its engine is asked again (ask_again, its event, or " + owner + "'s start)"}, nil + } + 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 +} diff --git a/go/clients/nodeengineclient/enginereads/enginereads_test.go b/go/clients/nodeengineclient/enginereads/enginereads_test.go new file mode 100644 index 0000000..b620d1c --- /dev/null +++ b/go/clients/nodeengineclient/enginereads/enginereads_test.go @@ -0,0 +1,40 @@ +package enginereads + +import ( + "strings" + "testing" +) + +// The answer and the event are judged as the contract lays them out and its reader checks: the layout's version, a node +// that is a name, no empty capability, a time, and an event's why one of the two. +func TestTheAnswerAndTheEventAreJudged(t *testing.T) { + const at = "2026-11-05T12:00:00Z" + for name, c := range map[string]struct{ body, bad string }{ + "whole": {`{"v":1,"node":"ace","manifest-reads":["mesh-config.resolved"],"stated-at":"` + at + `"}`, ""}, + "old layout": {`{"v":2,"node":"ace","stated-at":"` + at + `"}`, "version"}, + "bad node": {`{"v":1,"node":"a.b","stated-at":"` + at + `"}`, "no node"}, + "empty read": {`{"v":1,"node":"ace","manifest-reads":[" "],"stated-at":"` + at + `"}`, "empty reader"}, + "no time": {`{"v":1,"node":"ace","stated-at":"yesterday"}`, "RFC 3339"}, + } { + _, _, err := DecodeAnswer([]byte(c.body)) + if (c.bad == "") != (err == nil) || (err != nil && !strings.Contains(err.Error(), c.bad)) { + t.Errorf("answer %s: %v, want %q", name, err, c.bad) + } + } + for name, c := range map[string]struct{ body, bad string }{ + "whole": {`{"v":1,"node":"ace","why":"started","manifest-reads":["a.b"],"at":"` + at + `"}`, ""}, + "returned": {`{"v":1,"node":"ace","why":"returned","at":"` + at + `"}`, ""}, + "old layout": {`{"v":2,"node":"ace","why":"started","at":"` + at + `"}`, "version"}, + "no node": {`{"v":1,"node":"","why":"started","at":"` + at + `"}`, "no node"}, + "another why": {`{"v":1,"node":"ace","why":"bored","at":"` + at + `"}`, "bored"}, + "no time": {`{"v":1,"node":"ace","why":"started","at":"yesterday"}`, "RFC 3339"}, + } { + _, err := DecodeChanged([]byte(c.body)) + if (c.bad == "") != (err == nil) || (err != nil && !strings.Contains(err.Error(), c.bad)) { + t.Errorf("event %s: %v, want %q", name, err, c.bad) + } + } + if Verb != "manifest-reads" || Event != "manifest-reads-changed" || Consumes != "node-engine.manifest-reads-changed" || Invoke != "seat:node-engine.manifest-reads" { + t.Errorf("the names are %s %s %s %s", Verb, Event, Consumes, Invoke) + } +} diff --git a/health-and-repair/go/clients/meshconditionsclient/findings/findings.go b/health-and-repair/go/clients/meshconditionsclient/findings/findings.go new file mode 100644 index 0000000..903eb29 --- /dev/null +++ b/health-and-repair/go/clients/meshconditionsclient/findings/findings.go @@ -0,0 +1,88 @@ +// Package findings is a part's own bucket `findings` as the state the seat mesh-conditions' client writes into +// (meshconditionsclient.FindingState), so a part gives mesh-sdk's views the FindingWriter of the seat's layout without +// each writing the same adapter over its bucket (novox/hq ADR 0564 §2, ADR 0635 §4). +package findings + +import ( + "context" + "encoding/json" + "errors" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + "git.novox.be/novox/mesh-sdk/go/views" +) + +// ErrUnbound says the bucket is not on the bus yet, so the finding is not written (the part says it in its log). +var ErrUnbound = errors.New("the findings state is not bound yet") + +// State is the bucket the part's state `findings` is, given when it is bound (process.AwaitedBucket.KV): nil is not bound yet. +type State struct { + Ctx context.Context + KV func() jetstream.KeyValue +} + +var _ meshconditionsclient.FindingState = State{} + +// Put writes one value under its key as JSON. +func (s State) Put(key string, value any) (uint64, error) { + kv := s.KV() + if kv == nil { + return 0, ErrUnbound + } + data, err := json.Marshal(value) + if err != nil { + return 0, err + } + return kv.Put(s.Ctx, key, data) +} + +// Delete clears one key. +func (s State) Delete(key string) error { + kv := s.KV() + if kv == nil { + return ErrUnbound + } + return kv.Delete(s.Ctx, key) +} + +// Writer is the part's views.FindingWriter over its bucket; the function is what views.ProcessReader.Findings and the +// emitter's Findings take. +func Writer(ctx context.Context, kv func() jetstream.KeyValue) views.FindingWriter { + return meshconditionsclient.FindingWriter{State: State{Ctx: ctx, KV: kv}} +} + +// Funcs is a findings state held as the two functions a part keeps over it (the assigner's findings, which also say +// its conditions): a value is put as JSON. +type Funcs struct { + PutBytes func(key string, value []byte) error + DeleteKey func(key string) error +} + +var _ meshconditionsclient.FindingState = Funcs{} + +// Put writes one value under its key as JSON; the state not bound (no function) is ErrUnbound. +func (f Funcs) Put(key string, value any) (uint64, error) { + if f.PutBytes == nil { + return 0, ErrUnbound + } + data, err := json.Marshal(value) + if err != nil { + return 0, err + } + return 0, f.PutBytes(key, data) +} + +// Delete clears one key. +func (f Funcs) Delete(key string) error { + if f.DeleteKey == nil { + return ErrUnbound + } + return f.DeleteKey(key) +} + +// WriterOver is the part's views.FindingWriter over its two functions. +func WriterOver(put func(string, []byte) error, del func(string) error) views.FindingWriter { + return meshconditionsclient.FindingWriter{State: Funcs{PutBytes: put, DeleteKey: del}} +} diff --git a/health-and-repair/go/clients/meshconditionsclient/findings/findings_test.go b/health-and-repair/go/clients/meshconditionsclient/findings/findings_test.go new file mode 100644 index 0000000..01052ac --- /dev/null +++ b/health-and-repair/go/clients/meshconditionsclient/findings/findings_test.go @@ -0,0 +1,71 @@ +package findings + +import ( + "context" + "encoding/json" + "errors" + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/views" +) + +// fakeKV is the two calls a findings state makes of its bucket. +type fakeKV struct { + jetstream.KeyValue + held map[string][]byte +} + +func (f *fakeKV) Put(_ context.Context, key string, value []byte) (uint64, error) { + f.held[key] = value + return uint64(len(f.held)), nil +} + +func (f *fakeKV) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error { + delete(f.held, key) + return nil +} + +// A part's finding is written in the seat mesh-conditions' layout under the key the condition keeper reads, cleared by the +// same finding, and refused while the bucket is not bound (the part says that in its log). +func TestAFindingIsWrittenInTheSeatsLayoutAndClearedAndRefusedUnbound(t *testing.T) { + kv := &fakeKV{held: map[string][]byte{}} + var bound jetstream.KeyValue + w := Writer(context.Background(), func() jetstream.KeyValue { return bound }) + f := views.Finding{Observer: "mesh-delivery-teller", Kind: views.UnansweredKind, ID: "mesh-delivery-teller.asks", Summary: "s", Headline: "h", + Explanation: "e", At: time.Date(2026, 11, 5, 15, 0, 0, 0, time.UTC)} + if err := w.Put(f); !errors.Is(err, ErrUnbound) { + t.Fatalf("an unbound bucket took the finding: %v", err) + } + bound = kv + if err := w.Put(f); err != nil { + t.Fatal(err) + } + var got struct{ Key, Kind, Observer, Source string } + for key, value := range kv.held { + if err := json.Unmarshal(value, &got); err != nil || got.Key != key || got.Kind != views.UnansweredKind || + got.Observer != f.Observer || got.Source != f.Observer+"."+f.Kind { + t.Fatalf("%s holds %s (%v)", key, value, err) + } + } + if len(kv.held) != 1 { + t.Fatalf("the bucket holds %d findings, want 1", len(kv.held)) + } + if err := w.Delete(f); err != nil || len(kv.held) != 0 { + t.Fatalf("the finding is not cleared: %v, %d left", err, len(kv.held)) + } + // The same finding over two functions (a part whose findings also say its conditions). + held := map[string][]byte{} + o := WriterOver(func(k string, v []byte) error { held[k] = v; return nil }, func(k string) error { delete(held, k); return nil }) + if err := o.Put(f); err != nil || len(held) != 1 { + t.Fatalf("over functions: %v, %d held", err, len(held)) + } + if err := o.Delete(f); err != nil || len(held) != 0 { + t.Fatalf("over functions not cleared: %v, %d left", err, len(held)) + } + if err := WriterOver(nil, nil).Put(f); !errors.Is(err, ErrUnbound) { + t.Fatalf("no functions took the finding: %v", err) + } +} diff --git a/health-and-repair/go/clients/meshconditionsclient/go.mod b/health-and-repair/go/clients/meshconditionsclient/go.mod index 8843c1e..9e772c2 100644 --- a/health-and-repair/go/clients/meshconditionsclient/go.mod +++ b/health-and-repair/go/clients/meshconditionsclient/go.mod @@ -8,7 +8,7 @@ require git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000 require ( github.com/klauspost/compress v1.20.1 // indirect - github.com/nats-io/nats.go v1.54.0 // indirect + github.com/nats-io/nats.go v1.54.0 github.com/nats-io/nkeys v0.4.16 // indirect github.com/nats-io/nuid v1.0.1 // indirect golang.org/x/crypto v0.57.0 // indirect diff --git a/health-and-repair/go/clients/meshconditionsclient/heard/heard.go b/health-and-repair/go/clients/meshconditionsclient/heard/heard.go new file mode 100644 index 0000000..e00d535 --- /dev/null +++ b/health-and-repair/go/clients/meshconditionsclient/heard/heard.go @@ -0,0 +1,115 @@ +// Package heard is the finding a part writes while one of its consumers or watches does not hear the +// family it follows (novox/hq ADR 0615 §5; review round 1 of the registry milestone): the consumer not there yet, a +// hearing that could not start, a watch that cannot be opened. Without it the part's health stays healthy and its +// status says answered while the events it waits on are lost to it. The finding names the observer and the family, +// says what is not heard and why, and is deleted when the consumer binds: at the part's next start, or where the part +// binds again at a change of its membership, then; the condition keeper raises and clears the condition from it (ADR 0564 §2). +package heard + +import ( + "encoding/json" + "fmt" + "strings" + "time" + + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + "git.novox.be/novox/mesh-sdk/go/views" +) + +// Kind is the finding's kind. +const Kind = "family-not-heard" + +// ID is the finding's subject: `.`, the family by the name the part follows it under (a stream, an +// event, a state's watch). +func ID(observer, family string) string { return observer + "." + family } + +// Key is the finding's key in the observer's findings state. +func Key(observer, family string) string { + return meshconditionsclient.Key(meshconditionsclient.ScopeModule, ID(observer, family), Kind) +} + +// Words are the finding's words: its summary, headline and explanation. consequence says what the observer misses while +// it does not hear family ("the placement is the first view's alone"). +func Words(observer, family, consequence string, why error) (summary, headline, explanation string) { + reason := "it gave no reason" + if why != nil { + reason = strings.TrimSpace(why.Error()) + } + summary = fmt.Sprintf("%s does not hear %s: %s (%s)", observer, family, consequence, reason) + headline = fmt.Sprintf("%s does not hear %s", observer, family) + explanation = fmt.Sprintf("%s follows %s on a consumer of its own, and that consumer is not there or could not be heard, so %s. "+ + "This clears when the consumer binds: at %s's next start, or sooner where %s binds again at a change of its membership. "+ + "Look at whether the grants part made %s's consumers, then start %s again if it does not bind by itself.", observer, family, consequence, observer, observer, observer, observer) + return summary, headline, explanation +} + +// Finding is observer's finding that it does not hear family, in the layout of the seat mesh-conditions' state findings. +func Finding(observer, family, consequence string, why error, at time.Time) (string, []byte, error) { + summary, headline, explanation := Words(observer, family, consequence, why) + v := meshconditionsclient.FindingFor(views.Finding{Observer: observer, Kind: Kind, ID: ID(observer, family), Summary: summary, + Headline: headline, Explanation: explanation, At: at}) + if err := meshconditionsclient.CheckFinding(v); err != nil { + return v.Key, nil, err + } + value, err := json.Marshal(v) + return v.Key, value, err +} + +// Writer writes and deletes the finding in the observer's findings state. Put or Delete nil writes nothing, said in the +// log, so the part still says it. +type Writer struct { + Observer string + Put func(key string, value []byte) error + Delete func(key string) error + Logf func(string, ...any) + Now func() time.Time +} + +// Bind binds the observer's consumer of family (bind makes it and starts hearing it): the finding is deleted when it +// binds, and written, with why, when it does not. It answers bind's error, so a caller that cannot go on without the +// family still stops. +func (w Writer) Bind(family, consequence string, bind func() error) error { + if err := bind(); err != nil { + w.NotHeard(family, consequence, err) + return err + } + w.Heard(family) + return nil +} + +// NotHeard says in the log and in the finding that the observer does not hear family. +func (w Writer) NotHeard(family, consequence string, why error) { + now := time.Now + if w.Now != nil { + now = w.Now + } + summary, _, _ := Words(w.Observer, family, consequence, why) + w.logf("%s", summary) + if w.Put == nil { + w.logf("the finding %s is not written: no findings state", Key(w.Observer, family)) + return + } + key, value, err := Finding(w.Observer, family, consequence, why, now()) + if err == nil { + err = w.Put(key, value) + } + if err != nil { + w.logf("the finding %s is not written: %v", key, err) + } +} + +// Heard deletes the finding: the consumer of family is bound. +func (w Writer) Heard(family string) { + if w.Delete == nil { + return + } + if err := w.Delete(Key(w.Observer, family)); err != nil { + w.logf("the finding %s is not deleted: %v", Key(w.Observer, family), err) + } +} + +func (w Writer) logf(format string, args ...any) { + if w.Logf != nil { + w.Logf(format, args...) + } +} diff --git a/health-and-repair/go/clients/meshconditionsclient/heard/heard_test.go b/health-and-repair/go/clients/meshconditionsclient/heard/heard_test.go new file mode 100644 index 0000000..cc9b69a --- /dev/null +++ b/health-and-repair/go/clients/meshconditionsclient/heard/heard_test.go @@ -0,0 +1,36 @@ +package heard + +import ( + "errors" + "strings" + "testing" + "time" +) + +// A consumer that does not bind writes the finding, with why, under the observer's findings key; the one that binds +// deletes it; and a state that is not bound is said in the log, not written. +func TestBindWritesTheFindingUntilTheConsumerBinds(t *testing.T) { + puts := map[string][]byte{} + var logs []string + w := Writer{Observer: "mesh-x", Put: func(k string, v []byte) error { puts[k] = v; return nil }, + Delete: func(k string) error { delete(puts, k); return nil }, Logf: func(f string, a ...any) { logs = append(logs, f) }, + Now: func() time.Time { return time.Date(2026, 11, 5, 12, 0, 0, 0, time.UTC) }} + key := Key("mesh-x", "placement") + if want := "mesh-x.placement"; ID("mesh-x", "placement") != want || !strings.Contains(key, want) || !strings.Contains(key, Kind) { + t.Fatalf("the finding's key %q", key) + } + if err := w.Bind("placement", "the first view is the only picture", func() error { return errors.New("no consumer") }); err == nil { + t.Fatal("a failed bind was not answered") + } + if v := string(puts[key]); !strings.Contains(v, "no consumer") || !strings.Contains(v, "mesh-x does not hear placement") { + t.Errorf("the finding is %q", v) + } + if err := w.Bind("placement", "", func() error { return nil }); err != nil || len(puts) != 0 { + t.Errorf("a bound consumer left %v (%v)", puts, err) + } + w.Put = nil + w.NotHeard("placement", "x", nil) + if len(logs) < 2 || !strings.Contains(logs[len(logs)-1], "not written") { + t.Errorf("an unbound state was not said: %v", logs) + } +}