mesh-contracts/go/clients/nodeengineclient/enginereads/calls.go

package enginereads

import (
	"context"
	"maps"
	"slices"
	"strings"
	"sync"
	"time"
)

// CallID is the id of a part's finding about node's manifest-reads: the node the kept call waits on.
func CallID(node string) string { return "manifest-reads." + node }

// BoundPrefix starts the key of a node's bound in a part's state of bounds, a key that expires at its end.
const BoundPrefix = "reads-call."

// BoundOf is the node a bound key is of.
func BoundOf(key string) (string, bool) { return strings.CutPrefix(key, BoundPrefix) }

// UnansweredCall is a node's manifest-reads left unanswered past the bound while its machine is heard: what the part's
// finding call-unanswered says (novox/hq ADR 0668 §3), and At, the part's clock when it is written.
type UnansweredCall struct {
	Node  string
	Since time.Time
	Words string
	At    time.Time
}

// ReadsCalls is the wait of a part's kept manifest-reads calls through a Keeper, shared by every caller of the verb (the
// assigner, the builder, the indexer): a node whose engine has not answered for Bound while its machine is heard is
// Raised in the part's own findings, and Cleared once it answers (novox/hq ADR 0668 §3). Every step is an event; nothing
// waits on a timer.
//
// With Arm set, a node unanswered is weighed when its bound ends (Expired, on the expiry of the key Arm keeps in the
// part's state of bounds): Heard is asked then, once, and a node not heard has no finding; it is asked again when it
// returns, on manifest-reads-changed (ADR 0668 §4). With no Arm, Heard is the part's own knowledge, asked at each Step.
type ReadsCalls struct {
	Keeper *Keeper
	// Nodes are the counted nodes.
	Nodes func() []string
	// Heard says whether node's machine is heard (with Arm, the seat mesh-watchdogs' verb presence); an error is said and
	// reads as not heard.
	Heard func(ctx context.Context, node string) (bool, error)
	// Bound is how long a node may stay unanswered before it is the finding.
	Bound time.Duration
	// Raise writes the finding of u in the part's findings; Clear removes the finding of node, none being nothing to clear.
	Raise func(ctx context.Context, u UnansweredCall) error
	Clear func(ctx context.Context, node string) error
	// Held are the nodes whose finding the part's findings hold, written by this process or another; nil clears every
	// counted node not raised.
	Held func(ctx context.Context) ([]string, error)
	// Arm keeps a bound for node ending at ends, whose expiry is Expired; nil weighs Heard at each Step.
	Arm  func(ctx context.Context, node string, ends time.Time) error
	Now  func() time.Time
	Logf func(string, ...any)

	mu   sync.Mutex
	said map[string]time.Time // node -> the since its finding was written for
}

func (c *ReadsCalls) now() time.Time {
	if c.Now != nil {
		return c.Now()
	}
	return time.Now()
}

func (c *ReadsCalls) logf(f string, a ...any) {
	if c.Logf != nil {
		c.Logf(f, a...)
	}
}

func (c *ReadsCalls) ready() bool {
	return c != nil && c.Keeper != nil && c.Nodes != nil && c.Heard != nil && c.Raise != nil && c.Clear != nil
}

// unanswered is node's since while its statement is unanswered.
func unanswered(statements map[string]Statement, node string) (time.Time, bool) {
	s, ok := statements[node]
	if !ok || s.Answered || s.UnansweredSince.IsZero() {
		return time.Time{}, false
	}
	return s.UnansweredSince, true
}

func (c *ReadsCalls) heard(ctx context.Context, node string) bool {
	ok, err := c.Heard(ctx, node)
	if err != nil {
		c.logf("whether %s's machine is heard cannot be read, so its unanswered %s is no finding: %v", node, Verb, err)
		return false
	}
	return ok
}

func (c *ReadsCalls) isSaid(node string, since time.Time) bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	was, ok := c.said[node]
	return ok && was.Equal(since)
}

// raise writes node's finding, unless written for this since already, and asks the node again once its last ask ended.
func (c *ReadsCalls) raise(ctx context.Context, node string, since time.Time, words string) bool {
	if !c.Keeper.Asking(node) {
		c.Keeper.AskNode(node, "unanswered since "+since.UTC().Format(time.RFC3339)+", its machine heard")
	}
	if c.isSaid(node, since) {
		return true
	}
	if err := c.Raise(ctx, UnansweredCall{Node: node, Since: since, Words: words, At: c.now().UTC()}); err != nil {
		c.logf("the finding call-unanswered of %s's %s is not written: %v", node, Verb, err)
		return false
	}
	c.logf("%s on %s is unanswered since %s and its machine is heard: the finding call-unanswered", Verb, node, since.UTC().Format(time.RFC3339))
	c.mu.Lock()
	if c.said == nil {
		c.said = map[string]time.Time{}
	}
	c.said[node] = since
	c.mu.Unlock()
	return true
}

// Step weighs every counted node after a statement changed or the counted nodes changed: with no Arm a node unanswered
// and heard is asked again, and past Bound it is the finding; with Arm, a node unanswered and not yet said has its bound
// kept. Every finding no longer raised is cleared.
func (c *ReadsCalls) Step(ctx context.Context) {
	if !c.ready() {
		return
	}
	nodes := c.Nodes()
	statements := c.Keeper.Statements()
	now := c.now()
	raised := map[string]bool{}
	for _, node := range nodes {
		since, ok := unanswered(statements, node)
		if !ok {
			continue
		}
		if c.Arm != nil {
			if c.isSaid(node, since) {
				raised[node] = true
				continue
			}
			if err := c.Arm(ctx, node, since.Add(c.Bound)); err != nil {
				c.logf("the bound of %s's %s is not kept: %v", node, Verb, err)
			}
			continue
		}
		if !c.heard(ctx, node) {
			continue
		}
		if now.Sub(since) < c.Bound {
			if !c.Keeper.Asking(node) {
				c.Keeper.AskNode(node, "unanswered since "+since.UTC().Format(time.RFC3339)+", its machine heard")
			}
			continue
		}
		raised[node] = c.raise(ctx, node, since, statements[node].Words)
	}
	c.clear(ctx, nodes, raised)
}

// Expired weighs node when its bound ended: unanswered past Bound and its machine heard, it is the finding and is asked
// again; not yet past Bound, its bound is kept again; not heard, it has none.
func (c *ReadsCalls) Expired(ctx context.Context, node string) {
	if !c.ready() || !slices.Contains(c.Nodes(), node) {
		return
	}
	statements := c.Keeper.Statements()
	since, ok := unanswered(statements, node)
	if !ok || c.isSaid(node, since) {
		return
	}
	if c.now().Sub(since) < c.Bound {
		if c.Arm != nil {
			if err := c.Arm(ctx, node, since.Add(c.Bound)); err != nil {
				c.logf("the bound of %s's %s is not kept: %v", node, Verb, err)
			}
		}
		return
	}
	if !c.heard(ctx, node) {
		c.logf("%s on %s is unanswered since %s and its machine is not heard: no finding; it is asked again when it returns", Verb, node, since.UTC().Format(time.RFC3339))
		return
	}
	c.raise(ctx, node, since, statements[node].Words)
}

// clear removes the finding of every node held or said and not raised: with Held those it names, else every counted node.
func (c *ReadsCalls) clear(ctx context.Context, nodes []string, raised map[string]bool) {
	candidates := map[string]bool{}
	if c.Held != nil {
		held, err := c.Held(ctx)
		if err != nil {
			c.logf("the findings of %s held cannot be listed: %v", Verb, err)
		}
		for _, node := range held {
			candidates[node] = true
		}
	} else {
		for _, node := range nodes {
			candidates[node] = true
		}
	}
	c.mu.Lock()
	for node := range c.said {
		candidates[node] = true
	}
	c.mu.Unlock()
	for _, node := range slices.Sorted(maps.Keys(candidates)) {
		if raised[node] {
			continue
		}
		if err := c.Clear(ctx, node); err != nil {
			c.logf("the finding call-unanswered of %s's %s is not cleared: %v", node, Verb, err)
			continue
		}
		c.mu.Lock()
		delete(c.said, node)
		c.mu.Unlock()
	}
}