From bb16f99e2d5e854bb8f6552ce4b256e7c10e8ea0 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 10 Nov 2026 04:54:06 +0100 Subject: [PATCH] =?UTF-8?q?C7:=20the=20node-engine=20seat's=20client=20kee?= =?UTF-8?q?ps=20the=20wait=20of=20every=20caller's=20kept=20manifest-reads?= =?UTF-8?q?=20calls,=20ReadsCalls=20beside=20Keeper.Unanswered,=20its=20bo?= =?UTF-8?q?und's=20end=20asking=20once=20whether=20the=20machine=20is=20he?= =?UTF-8?q?ard=20(issue=201966,=20ADR=200668=20=C2=A73,=20ADR=200634=20?= =?UTF-8?q?=C2=A71,=20ADR=200596=20=C2=A72;=20review=20round=203,=20findin?= =?UTF-8?q?gs=201=20and=202)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../nodeengineclient/enginereads/calls.go | 225 ++++++++++++++++++ .../enginereads/enginereads_test.go | 68 ++++++ 2 files changed, 293 insertions(+) create mode 100644 go/clients/nodeengineclient/enginereads/calls.go diff --git a/go/clients/nodeengineclient/enginereads/calls.go b/go/clients/nodeengineclient/enginereads/calls.go new file mode 100644 index 0000000..dab4040 --- /dev/null +++ b/go/clients/nodeengineclient/enginereads/calls.go @@ -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() + } +} diff --git a/go/clients/nodeengineclient/enginereads/enginereads_test.go b/go/clients/nodeengineclient/enginereads/enginereads_test.go index 65f8c91..607e3d0 100644 --- a/go/clients/nodeengineclient/enginereads/enginereads_test.go +++ b/go/clients/nodeengineclient/enginereads/enginereads_test.go @@ -99,3 +99,71 @@ func TestTheAnswerAndTheEventAreJudged(t *testing.T) { 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) + } +}