C7: the node-engine seat's client keeps the wait of every caller's kept manifest-reads calls, ReadsCalls beside Keeper.Unanswered, its bound's end asking once whether the machine is heard (issue 1966, ADR 0668 §3, ADR 0634 §1, ADR 0596 §2; review round 3, findings 1 and 2)
The assigner, the builder and the indexer share it instead of two copies; it names no other seat's client (the client check), so each caller writes its finding in mesh-conditions' layout through Raise, stamped with ReadsCalls' clock. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -0,0 +1,225 @@
|
|||||||
|
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()
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -99,3 +99,71 @@ func TestTheAnswerAndTheEventAreJudged(t *testing.T) {
|
|||||||
t.Errorf("the names are %s %s %s %s", Verb, Event, Consumes, Invoke)
|
t.Errorf("the names are %s %s %s %s", Verb, Event, Consumes, Invoke)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ADR 0668 §3, ADR 0596 §2 (review round 3): the wait of a part's kept manifest-reads calls. With Arm, a node unanswered
|
||||||
|
// has its bound kept, and only its expiry asks whether its machine is heard, once: heard, the finding is written at the
|
||||||
|
// part's clock and the node asked again; away, none; answered, it is cleared. With no Arm, heard is weighed at each step.
|
||||||
|
func TestReadsCallsSayAnUnansweredNodeHeard(t *testing.T) {
|
||||||
|
clock := time.Date(2026, 11, 10, 3, 0, 0, 0, time.UTC)
|
||||||
|
answers, calls := false, 0
|
||||||
|
k := &Keeper{Sync: true, Now: func() time.Time { return clock }, Call: func(node, _ string, _ any) (json.RawMessage, error) {
|
||||||
|
calls++
|
||||||
|
if answers {
|
||||||
|
return json.RawMessage(`{"v":1,"node":"` + node + `","manifest-reads":["a.b"],"stated-at":"2026-11-10T03:00:00Z"}`), nil
|
||||||
|
}
|
||||||
|
return nil, context.DeadlineExceeded
|
||||||
|
}}
|
||||||
|
for name, armed := range map[string]bool{"armed": true, "by its own knowledge": false} {
|
||||||
|
answers, calls = false, 0
|
||||||
|
k.Forget("ace")
|
||||||
|
first := clock
|
||||||
|
k.AskNode("ace", "the first view")
|
||||||
|
heard, asked := true, 0
|
||||||
|
raised, cleared := map[string]UnansweredCall{}, map[string]int{}
|
||||||
|
bounds := map[string]time.Time{}
|
||||||
|
c := &ReadsCalls{Keeper: k, Nodes: func() []string { return []string{"ace", "g14"} }, Bound: 5 * time.Minute,
|
||||||
|
Now: func() time.Time { return clock.Add(time.Second) },
|
||||||
|
Heard: func(context.Context, string) (bool, error) { asked++; return heard, nil },
|
||||||
|
Raise: func(_ context.Context, u UnansweredCall) error { raised[u.Node] = u; return nil },
|
||||||
|
Clear: func(_ context.Context, node string) error { cleared[node]++; delete(raised, node); return nil }}
|
||||||
|
if armed {
|
||||||
|
c.Arm = func(_ context.Context, node string, ends time.Time) error { bounds[node] = ends; return nil }
|
||||||
|
}
|
||||||
|
c.Step(context.Background())
|
||||||
|
if armed && (asked != 0 || !bounds["ace"].Equal(first.Add(5*time.Minute)) || len(raised) != 0) {
|
||||||
|
t.Errorf("%s: a step before the bound asked heard %d times, bounds %v, raised %v", name, asked, bounds, raised)
|
||||||
|
}
|
||||||
|
clock = clock.Add(6 * time.Minute)
|
||||||
|
heard, asked = false, 0
|
||||||
|
if armed {
|
||||||
|
c.Expired(context.Background(), "ace")
|
||||||
|
} else {
|
||||||
|
c.Step(context.Background())
|
||||||
|
}
|
||||||
|
if len(raised) != 0 || asked != 1 {
|
||||||
|
t.Errorf("%s: away, raised %v, heard asked %d times", name, raised, asked)
|
||||||
|
}
|
||||||
|
heard, asked, calls = true, 0, 0
|
||||||
|
if armed {
|
||||||
|
c.Expired(context.Background(), "ace")
|
||||||
|
} else {
|
||||||
|
c.Step(context.Background())
|
||||||
|
}
|
||||||
|
if u, ok := raised["ace"]; !ok || !u.Since.Equal(first) || !u.At.Equal(clock.Add(time.Second)) || asked != 1 || calls != 1 {
|
||||||
|
t.Errorf("%s: heard past the bound, raised %+v, heard asked %d, called %d", name, raised, asked, calls)
|
||||||
|
}
|
||||||
|
answers = true
|
||||||
|
k.AskNode("ace", "answered")
|
||||||
|
c.Step(context.Background())
|
||||||
|
if _, ok := raised["ace"]; ok || cleared["ace"] == 0 {
|
||||||
|
t.Errorf("%s: answered, raised %v, cleared %v", name, raised, cleared)
|
||||||
|
}
|
||||||
|
clock = clock.Add(time.Hour)
|
||||||
|
}
|
||||||
|
if CallID("ace") != "manifest-reads.ace" {
|
||||||
|
t.Errorf("the call's id is %s", CallID("ace"))
|
||||||
|
}
|
||||||
|
if n, ok := BoundOf(BoundPrefix + "ace"); !ok || n != "ace" {
|
||||||
|
t.Errorf("the bound's node is %q %v", n, ok)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user