Files
mesh-contracts/go/clients/nodeengineclient/enginereads/enginereads.go
T
jschoubben c9ffed7042
mesh/delivery merged: source.main moved to novox/mesh-contracts@f5d76e84ca71
mesh/delivery-group group fix/1966-engine-reads-asked-again merged: every member is merged
mesh/check checking c9ffed70421a on f5d76e84ca71 | novox/mesh-contracts#4 in group fix/1966-engine-reads-asked-again
S1-S3: the node-engine seat's client asks a node once more when an ask came while one ran, and shows when the event was heard (issue 1966 note 170)
An event of manifest-reads-changed heard while an ask of the same node ran was dropped
without a line, so a restated engine could stay read at its first statement (ace,
2026-11-10 07:36). Every reader goes through Keeper.AskNode, so the assigner, the
builder and the indexer are fixed at once (ADR 0668 §4: the caller hears it and calls
again; ADR 0634 §1: the seat's client).

- AskNode: an ask while one runs is asked once more after it, one follow-up however
  many came, said in one line (1966/AC170-1, AC170-3). The asks of one node so run one
  after another, and take keeps the last answer, as before (no rule on the engine's
  clock: review round 1).
- Statement.EventAt (event-at): when the node's event was last heard, read by the
  parts' engine-reads tools; forgotten with the node's key.

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
2026-11-10 11:42:05 +01:00

755 lines
27 KiB
Go

// 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>.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"`
// UnansweredSince is when the engine first did not answer, of the asks since its last answer; zero once answered
// (ADR 0668 §3: a kept call unanswered past five minutes is a finding of its caller).
UnansweredSince time.Time `json:"unanswered-since,omitzero"`
// EventAt is when the part last heard the seat's event manifest-reads-changed for the node, beside when it asked and
// what the engine stated then (issue 1966 note 170: was the event heard, and was it asked again after it).
EventAt time.Time `json:"event-at,omitzero"`
}
// 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.CallKeptOn, a kept call asked again through a transport fault
// within the bus's retry bound (novox/hq ADR 0668 §1; the verb is a read, safe to call again).
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
// Once asks the verb once (process.Conn.CallOn). When set, its fault keeps the fallback or the last statement at once,
// marked as not answering, while Call, the kept call, asks on until answered (novox/hq ADR 0668 §3): a part does not
// wait out busretry.Bound for a node whose engine answers nothing before it acts on what it last knew.
Once Caller
// Fallback reads a node's record `<node>.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
// again is, per node an ask of which runs, the newest why of an ask that came meanwhile: it is asked once more when
// the running ask ends, never dropped (issue 1966 note 170).
again map[string]string
// eventAt is, per node, when its event manifest-reads-changed was last heard (Statement.EventAt).
eventAt map[string]time.Time
}
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; why is said in the log. When an ask of it runs already, the node is asked once more after
// that ask ends, one follow-up however many asks came meanwhile, so an event heard while an ask runs is never lost (issue
// 1966 note 170; ADR 0668 §4: the caller hears it and calls again). A why that starts with Event is the seat's event,
// whose time the node's statement shows as event-at.
func (k *Keeper) AskNode(node, why string) {
k.mu.Lock()
if k.asking == nil {
k.asking, k.again, k.eventAt = map[string]bool{}, map[string]string{}, map[string]time.Time{}
}
if strings.HasPrefix(why, Event) {
k.eventAt[node] = k.now().UTC()
}
if k.asking[node] {
_, waits := k.again[node]
k.again[node] = why
k.mu.Unlock()
if !waits {
k.logf("%s on %s: an ask runs, so it is asked once more after it (%s)", Verb, node, why)
}
return
}
k.asking[node] = true
k.mu.Unlock()
if k.Sync {
k.ask(node, why)
return
}
go k.ask(node, why)
}
// AskAll asks every node of nodes once, each as AskNode does.
func (k *Keeper) AskAll(nodes []string, why string) {
for _, n := range nodes {
k.AskNode(n, why)
}
}
// AtBind is what a part registers with its event hub's OnBound (process.EventHub.OnBound): at each bind of its consumer
// every node nodes names is asked once, so a manifest-reads-changed said before the consumer existed, and a statement
// taken while an engine restarted, end at the bind and not at the next event (novox/hq ADR 0668 §4; issue 1966 note 155).
func (k *Keeper) AtBind(nodes func() []string) func(created time.Time) {
return func(created time.Time) {
k.AskAll(nodes(), "the part's consumer was bound, made "+created.UTC().Format(time.RFC3339))
}
}
// ask asks node, and once more for as long as an ask came while it ran (AskNode); only then is the node no longer asking.
func (k *Keeper) ask(node, why string) {
for {
k.askOnce(node, why)
k.mu.Lock()
next, again := k.again[node]
delete(k.again, node)
if !again {
delete(k.asking, node)
}
k.mu.Unlock()
if !again {
return
}
why = "asked again: " + next
}
}
func (k *Keeper) askOnce(node, why string) {
if k.Call != nil && k.Once != nil {
first := Ask(k.Once, node, k.now())
if first.Answered {
k.take(first)
return
}
k.settle(first, why)
why += "; the kept call asks on after: " + first.Words
}
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())
}
k.settle(s, why)
}
// settle takes s, read from the node's record in its place where the engine did not answer.
func (k *Keeper) settle(s Statement, why string) {
node := s.Node
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)
}
// Unanswered is, of nodes, each whose engine the kept call has not answered for bound or longer by now while heard says
// its machine is heard, with since when (novox/hq ADR 0668 §3); a node whose machine is away is none. With no Call, none.
func (k *Keeper) Unanswered(nodes []string, heard func(node string) bool, bound time.Duration, now time.Time) map[string]time.Time {
out := map[string]time.Time{}
if k.Call == nil {
return out
}
k.mu.Lock()
defer k.mu.Unlock()
for _, node := range nodes {
s, ok := k.by[node]
if ok && !s.Answered && !s.UnansweredSince.IsZero() && now.Sub(s.UnansweredSince) >= bound && heard(node) {
out[node] = s.UnansweredSince
}
}
return out
}
// Asking says an ask of node runs now.
func (k *Keeper) Asking(node string) bool {
k.mu.Lock()
defer k.mu.Unlock()
return k.asking[node]
}
// 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. The asks of one node run one after another (AskNode), so the last answer is the newest.
// The statement shows when the node's event was last heard.
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 && old.EventAt.After(s.EventAt) {
s.EventAt = old.EventAt
}
if at := k.eventAt[s.Node]; at.After(s.EventAt) {
s.EventAt = at
}
if !s.Answered {
s.UnansweredSince = s.AskedAt
if old, ok := k.by[s.Node]; ok && !old.Answered && !old.UnansweredSince.IsZero() {
s.UnansweredSince = old.UnansweredSince
}
}
if old, ok := k.by[s.Node]; ok && !s.Answered && old.From != FromNone {
old.Answered, old.AskedAt, old.Words, old.AskAgain, old.UnansweredSince = false, s.AskedAt, s.Words, nil, s.UnansweredSince
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)
delete(k.eventAt, 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)
delete(k.eventAt, 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
}