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() } }