diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index 6f067139..31cebd7b 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -43,8 +43,8 @@ type passer struct { said map[string]bool // findings is the assigner's findings bucket (its said conditions, ADR 0564 §2). findings assign.Findings - // unansweredSaid are the call-unanswered findings of manifest-reads this process wrote (ADR 0668 §3). - unansweredSaid map[string]bool + // calls is the wait of the kept manifest-reads calls (enginereads.ReadsCalls, ADR 0668 §3), made at the first pass. + calls *enginereads.ReadsCalls // failedProvisions are the failed provisions whose finding this process wrote (ADR 0570 §4). failedProvisions map[string]bool // missing are the required provisions with no key this process pulsed, and missingSaid those whose finding it wrote. @@ -122,69 +122,59 @@ func (p *passer) engineReads(n assign.Nodes) { p.unansweredReads(known, n.Online) } -// readsCallID is the finding's id for a node's manifest-reads: the node the kept call waits on. -func readsCallID(node string) string { return "manifest-reads." + node } - // unansweredReads keeps the wait of every node whose engine has not answered manifest-reads while its machine is heard -// (online): it is asked again once its last ask ended, so the kept call goes on until answered; past +// (online), through the node-engine seat's client (enginereads.ReadsCalls, shared with the builder and the indexer): it +// is asked again once its last ask ended, so the kept call goes on until answered; past // meshconditionsclient.CallUnansweredAfter it is the finding call-unanswered in the assigner's findings, cleared once -// answered or while its machine is away, when none is written (novox/hq ADR 0668 §3). Each pass is the event: no timer. +// answered (novox/hq ADR 0668 §3). Each pass is the event, and the assigner's own presence says heard: no timer, no bound. func (p *passer) unansweredReads(known []string, online map[string]bool) { if p.reads.Call == nil { return } - heard := func(node string) bool { return online[node] } - for _, node := range known { - if s, ok := p.reads.Statements()[node]; ok && !s.Answered && heard(node) && !p.reads.Asking(node) { - p.reads.AskNode(node, "unanswered since "+s.UnansweredSince.UTC().Format(time.RFC3339)+", its machine heard") - } - } if p.findings.Put == nil || p.findings.Delete == nil { + for _, node := range known { + if s, ok := p.reads.Statements()[node]; ok && !s.Answered && online[node] && !p.reads.Asking(node) { + p.reads.AskNode(node, "unanswered since "+s.UnansweredSince.UTC().Format(time.RFC3339)+", its machine heard") + } + } return } - now := p.clock() - statements := p.reads.Statements() - raised := map[string]bool{} - for node, since := range p.reads.Unanswered(known, heard, mcc.CallUnansweredAfter, now) { - o := mcc.CallUnanswered(readsCallID(node), enginereads.Invoke, node, statements[node].Words, "assigner."+mcc.KindCallUnanswered, since) - o.Needs = "assigner_engine_reads with ask_again asks the node again now and shows its statement." - key := o.Key() - raised[key] = true - if p.unansweredSaid[key] { - continue - } - value, err := mcc.EncodeFinding(o.Finding(plc.Module, "", now.UTC())) - if err == nil { - err = p.findings.Put(key, value) - } - if err != nil { - p.logf("the finding %s is not written: %v", key, err) - continue - } - p.logf("%s on %s is unanswered since %s and its machine is heard: the finding %s", enginereads.Verb, node, since.UTC().Format(time.RFC3339), key) - if p.unansweredSaid == nil { - p.unansweredSaid = map[string]bool{} - } - p.unansweredSaid[key] = true - } - // What a pass before or another process wrote: every such finding of the assigner's no longer raised is cleared. - prefix, suffix := mcc.ScopeCall+"."+readsCallID(""), "."+mcc.KindCallUnanswered - keys := slices.Collect(maps.Keys(p.unansweredSaid)) - if p.findings.Keys != nil { - if held, err := p.findings.Keys(); err == nil { - keys = append(keys, held...) + if p.calls == nil || p.calls.Keeper != p.reads { + p.calls = &enginereads.ReadsCalls{Keeper: p.reads, Bound: mcc.CallUnansweredAfter, Now: p.clock, Logf: p.logf, + Raise: func(_ context.Context, u enginereads.UnansweredCall) error { + o := mcc.CallUnanswered(enginereads.CallID(u.Node), enginereads.Invoke, u.Node, u.Words, "assigner."+mcc.KindCallUnanswered, u.Since) + o.Needs = "assigner_engine_reads with ask_again asks the node again now and shows its statement." + value, err := mcc.EncodeFinding(o.Finding(plc.Module, "", u.At)) + if err == nil { + err = p.findings.Put(o.Key(), value) + } + return err + }, + Clear: func(_ context.Context, node string) error { + return p.findings.Delete(mcc.Key(mcc.ScopeCall, enginereads.CallID(node), mcc.KindCallUnanswered)) + }} + if p.findings.Keys != nil { + // What a pass before or another process wrote: every such finding of the assigner's is weighed to be cleared. + prefix, suffix := mcc.ScopeCall+"."+enginereads.CallID(""), "."+mcc.KindCallUnanswered + p.calls.Held = func(context.Context) ([]string, error) { + keys, err := p.findings.Keys() + var nodes []string + for _, key := range keys { + if rest, ok := strings.CutPrefix(key, prefix); ok { + if node, ok := strings.CutSuffix(rest, suffix); ok { + nodes = append(nodes, node) + } + } + } + return nodes, err + } + } else { + p.calls.Held = func(context.Context) ([]string, error) { return nil, nil } } } - for _, key := range keys { - if raised[key] || !strings.HasPrefix(key, prefix) || !strings.HasSuffix(key, suffix) { - continue - } - if err := p.findings.Delete(key); err != nil { - p.logf("the finding %s is not cleared: %v", key, err) - continue - } - delete(p.unansweredSaid, key) - } + p.calls.Nodes = func() []string { return known } + p.calls.Heard = func(_ context.Context, node string) (bool, error) { return online[node], nil } + p.calls.Step(context.Background()) } func (p *passer) clock() time.Time { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index 50cba6c4..b8e5bcbd 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -287,7 +287,7 @@ func TestAnAnswerCountsOnlyForTheAskingItAnswered(t *testing.T) { Delete: func(k string) error { delete(held, k); return nil }, Keys: func() ([]string, error) { return slices.Collect(maps.Keys(held)), nil }}, reads: &enginereads.Keeper{Call: silent, Sync: true, Now: func() time.Time { return clock }}} - key := mcc.Key(mcc.ScopeCall, readsCallID("ace"), mcc.KindCallUnanswered) + key := mcc.Key(mcc.ScopeCall, enginereads.CallID("ace"), mcc.KindCallUnanswered) before := asked["ace"] q.engineReads(w.nodes(nil, nil)) clock = clock.Add(6 * time.Minute)