mesh-delivery/sdk/enginereads/enginereads.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 mesh-delivery's, not mesh-sdk's (ADR 0593 §1, ADR 0608 §5).
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"`
}

// 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>.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
}