diff --git a/go/clients/nodeengineclient/enginereads/enginereads.go b/go/clients/nodeengineclient/enginereads/enginereads.go index 4ae20b2..c86476b 100644 --- a/go/clients/nodeengineclient/enginereads/enginereads.go +++ b/go/clients/nodeengineclient/enginereads/enginereads.go @@ -170,7 +170,8 @@ func Decode(raw []byte) (Statement, error) { return s, nil } -// Caller calls a seat's verb on one node: process.Conn.CallOn. +// Caller calls a seat's verb on one node: process.Conn.CallKeptOn, a kept call asked again through a transport fault +// within the bus's retry bound (novox/hq ADR 0668 §1; the verb is a read, safe to call again). 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 @@ -269,6 +270,22 @@ func (k *Keeper) AskNode(node, why string) { go k.ask(node, why) } +// AskAll asks every node of nodes once, each as AskNode does. +func (k *Keeper) AskAll(nodes []string, why string) { + for _, n := range nodes { + k.AskNode(n, why) + } +} + +// AtBind is what a part registers with its event hub's OnBound (process.EventHub.OnBound): at each bind of its consumer +// every node nodes names is asked once, so a manifest-reads-changed said before the consumer existed, and a statement +// taken while an engine restarted, end at the bind and not at the next event (novox/hq ADR 0668 §4; issue 1966 note 155). +func (k *Keeper) AtBind(nodes func() []string) func(created time.Time) { + return func(created time.Time) { + k.AskAll(nodes(), "the part's consumer was bound, made "+created.UTC().Format(time.RFC3339)) + } +} + func (k *Keeper) ask(node, why string) { defer func() { k.mu.Lock() diff --git a/go/clients/nodeengineclient/enginereads/enginereads_test.go b/go/clients/nodeengineclient/enginereads/enginereads_test.go index b620d1c..135c660 100644 --- a/go/clients/nodeengineclient/enginereads/enginereads_test.go +++ b/go/clients/nodeengineclient/enginereads/enginereads_test.go @@ -1,8 +1,11 @@ package enginereads import ( + "context" + "encoding/json" "strings" "testing" + "time" ) // The answer and the event are judged as the contract lays them out and its reader checks: the layout's version, a node @@ -34,6 +37,24 @@ func TestTheAnswerAndTheEventAreJudged(t *testing.T) { t.Errorf("event %s: %v, want %q", name, err, c.bad) } } + // Issue 1966 note 155, ADR 0668 §4: a part whose first call of a node timed out, and whose consumer was bound after + // the node's event was said, calls every node once at the bind, and the node is then held as it answered. + calls := map[string]int{} + k := &Keeper{Sync: true, Call: func(node, _ string, _ any) (json.RawMessage, error) { + calls[node]++ + if node == "ace" && calls[node] == 1 { + return nil, context.DeadlineExceeded + } + return json.RawMessage(`{"v":1,"node":"` + node + `","manifest-reads":["a.b"],"stated-at":"` + at + `"}`), nil + }} + k.AskNode("ace", "the first view") + if k.Statements()["ace"].Answered { + t.Errorf("a call that timed out reads as answered") + } + k.AtBind(func() []string { return []string{"ace", "g14"} })(time.Unix(0, 0)) + if p := k.Present([]string{"ace", "g14"}); len(p["ace"]) != 1 || len(p["g14"]) != 1 || calls["ace"] != 2 || calls["g14"] != 1 { + t.Errorf("after the bind: present %v, calls %v", p, calls) + } if Verb != "manifest-reads" || Event != "manifest-reads-changed" || Consumes != "node-engine.manifest-reads-changed" || Invoke != "seat:node-engine.manifest-reads" { t.Errorf("the names are %s %s %s %s", Verb, Event, Consumes, Invoke) }