diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/askended.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/askended.go index 893232ee..7b041c7f 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/askended.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/askended.go @@ -2,6 +2,7 @@ package main import ( "context" + "encoding/json" "errors" "time" @@ -9,11 +10,15 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-teller/tellerevents" + "git.novox.be/novox/mesh-sdk/go/busretry" "git.novox.be/novox/mesh-sdk/go/process" + mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "mesh-delivery-assigner/internal/assign" + plc "mesh-delivery-assigner/internal/assignment" "mesh-delivery-assigner/internal/buckets" "mesh-delivery-assigner/internal/puts" ) @@ -50,21 +55,71 @@ const channelAsksVerb = "seat:" + oc.Seat + "." + oc.VerbAsks // assigner's consumer: an ask-ended the teller said before the consumer existed is not delivered to it (issue 1966 note 155, // ADR 0668 §4), so each held ask whose asking ended there expired or cancelled is written again as that event writes it // (askAgain, with every rule and its compare-and-set). The read is a kept call (CallKept), made off the hub's start. -func readEndedAtBind(ctx context.Context, js jetstream.JetStream, hub *process.EventHub, asksKV jetstream.KeyValue, bus caller, logf func(string, ...any)) { +func readEndedAtBind(ctx context.Context, js jetstream.JetStream, hub *process.EventHub, asksKV jetstream.KeyValue, bus caller, wait callWait, logf func(string, ...any)) { if asksKV == nil || bus == nil { return } hub.OnBound(func(created time.Time) { go func() { - if err := endedAtBind(ctx, js, asksKV, bus, created, time.Now(), logf); err != nil { + if err := endedAtBind(ctx, js, asksKV, bus, wait, created, time.Now(), logf); err != nil { logf("the operator channel's ended asks are not read at the bind of %s (%v): an ask that ended there before is asked again when the teller says it ended", created.Format(time.RFC3339), err) } }() }) } -// endedAtBind is readEndedAtBind's one read: the held asks, the seat's verb, and each ended asking written again. -func endedAtBind(ctx context.Context, js jetstream.JetStream, asksKV jetstream.KeyValue, bus caller, created, now time.Time, logf func(string, ...any)) error { +// callWait is how a kept call the assigner waits on is said (novox/hq ADR 0668 §3): heard says the callee's machine is +// heard, now is the clock, and findings the assigner's findings, where the call unanswered past +// meshconditionsclient.CallUnansweredAfter is the finding call-unanswered while heard, cleared when it is answered. +type callWait struct { + heard func() bool + now func() time.Time + findings assign.Findings +} + +// asksCallID is the finding's id for the bind's read of the operator channel's asks. +const asksCallID = "operator-channel-asks" + +// keep makes call again after every transport fault (busretry: a request that reached no one, or one whose deadline +// passed) until it is answered or ctx ends, each try itself a kept call; past the bound with the callee heard it writes +// the finding call-unanswered, and clears it when answered. No timer: each try is the wait. +func (cw callWait) keep(ctx context.Context, id, callee string, call func() (json.RawMessage, error), logf func(string, ...any)) (json.RawMessage, error) { + clock := cw.now + if clock == nil { + clock = time.Now + } + since, said := clock(), false + key := mcc.Key(mcc.ScopeCall, id, mcc.KindCallUnanswered) + for { + answer, err := call() + if err == nil || !(busretry.Undelivered(err) || busretry.TimedOut(err)) || ctx.Err() != nil { + if said && cw.findings.Delete != nil { + if derr := cw.findings.Delete(key); derr != nil { + logf("the finding %s is not cleared: %v", key, derr) + } + } + return answer, err + } + if said || clock().Sub(since) < mcc.CallUnansweredAfter || cw.heard == nil || !cw.heard() || cw.findings.Put == nil { + continue + } + o := mcc.CallUnanswered(id, callee, "", err.Error(), "assigner."+mcc.KindCallUnanswered, since) + value, eerr := mcc.EncodeFinding(o.Finding(plc.Module, "", clock().UTC())) + if eerr == nil { + eerr = cw.findings.Put(key, value) + } + if eerr != nil { + logf("the finding %s is not written: %v", key, eerr) + continue + } + said = true + logf("%s is unanswered since %s and its callee's machine is heard: the finding %s", callee, since.UTC().Format(time.RFC3339), key) + } +} + +// endedAtBind is readEndedAtBind's one read: the held asks, the seat's verb (a kept call whose wait is kept, wait), and +// each ended asking written again. +func endedAtBind(ctx context.Context, js jetstream.JetStream, asksKV jetstream.KeyValue, bus caller, wait callWait, created, now time.Time, logf func(string, ...any)) error { keys, err := asksKV.Keys(ctx) if errors.Is(err, jetstream.ErrNoKeysFound) { return nil @@ -82,7 +137,7 @@ func endedAtBind(ctx context.Context, js jetstream.JetStream, asksKV jetstream.K if !ok { return nil } - answer, err := bus.CallKept(channelAsksVerb, query) + answer, err := wait.keep(ctx, asksCallID, channelAsksVerb, func() (json.RawMessage, error) { return bus.CallKept(channelAsksVerb, query) }, logf) if err != nil { return err } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 9fb35753..a18d9632 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -484,7 +484,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, if err := hearAskEnded(ctx, js, hub, kvs[buckets.Asks], logger.Printf); err != nil { logger.Printf("%s is not heard, so an ask that ends at the operator channel without an answer is not asked again until the assigner starts again: %v", askEndedSubject, err) } - readEndedAtBind(ctx, js, hub, kvs[buckets.Asks], bus, logger.Printf) + readEndedAtBind(ctx, js, hub, kvs[buckets.Asks], bus, callWait{heard: func() bool { return w.seatHeard(oc.Seat) }, findings: findings}, logger.Printf) hearReadsChanged(ctx, js, hub, reads, findings, gate.eventsLive, logger.Printf) } if bus != nil && kvs[buckets.Puts] != nil { 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 4dfc33e8..6f067139 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -24,6 +24,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads" + mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "mesh-delivery-assigner/internal/assign" plc "mesh-delivery-assigner/internal/assignment" @@ -42,6 +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 // 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. @@ -116,6 +119,72 @@ func (p *passer) engineReads(n assign.Nodes) { n.Reads[node] = reads } } + 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 +// 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. +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 { + 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...) + } + } + 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) + } } 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 64b17792..50cba6c4 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 @@ -7,6 +7,7 @@ import ( "errors" "fmt" stdio "git.novox.be/novox/mesh-sdk/go" + "maps" "mesh-delivery-assigner/internal/buckets" "mesh-delivery-assigner/internal/puts" "slices" @@ -36,6 +37,7 @@ import ( "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-delivery/sdk/graph" "mesh-delivery-assigner/internal/assign" @@ -153,6 +155,33 @@ func TestAHoldAskedAgainAfterALiftAndARestartHasANewID(t *testing.T) { t.Errorf("%s: asked again", c.why) } } + // ADR 0668 §3 (review round 1): the bind's read of the operator channel's asks keeps its wait through transport faults; + // past five minutes with the channel's machine heard it is the finding call-unanswered, cleared once answered; with + // the machine away none is written. A refusal is an answer and ends the wait. + for _, heard := range []bool{false, true} { + clock, findings := now, map[string][]byte{} + cw := callWait{heard: func() bool { return heard }, now: func() time.Time { return clock }, + findings: assign.Findings{Put: func(k string, v []byte) error { findings[k] = v; return nil }, + Delete: func(k string) error { delete(findings, k); return nil }}} + tries, during := 0, 0 + answer, err := cw.keep(context.Background(), asksCallID, channelAsksVerb, func() (json.RawMessage, error) { + tries++ + clock = clock.Add(2 * time.Minute) + if tries == 5 { + during = len(findings) + return json.RawMessage(`{"asks":[]}`), nil + } + return nil, errors.New("nats: no responders available for request") + }, t.Logf) + if want := map[bool]int{false: 0, true: 1}[heard]; err != nil || string(answer) != `{"asks":[]}` || tries != 5 || during != want || len(findings) != 0 { + t.Errorf("heard %v: %s %v after %d tries; findings during %d, after %d", heard, answer, err, tries, during, len(findings)) + } + } + if _, err := (callWait{}).keep(context.Background(), asksCallID, channelAsksVerb, func() (json.RawMessage, error) { + return nil, &busretry.Answered{Words: "refused"} + }, t.Logf); err == nil { + t.Error("a refusal was asked again") + } } // 1906/H4: an answer counts only for the asking it answered: an answer to an earlier hold of the same fingerprint, or to an ask @@ -240,6 +269,45 @@ func TestAnAnswerCountsOnlyForTheAskingItAnswered(t *testing.T) { if asked["ace"] != 2 { t.Errorf("ace, back online, was asked %d times, want 2", asked["ace"]) } + // ADR 0668 §3 (review round 1): a node whose engine answers nothing while its machine is heard is asked again at each + // pass once its ask ended, and past five minutes is the finding call-unanswered, cleared once it answers; none is + // written while its machine is away. + clock := time.Date(2026, 11, 10, 3, 0, 0, 0, time.UTC) + held := map[string][]byte{} + answers := false + silent := func(node, _ string, _ any) (json.RawMessage, error) { + asked[node]++ + if answers { + return json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: node, ManifestReads: []string{"form:new"}, StatedAt: clock.Format(time.RFC3339Nano)}) + } + return nil, context.DeadlineExceeded + } + q := &passer{w: w, logf: t.Logf, now: func() time.Time { return clock }, findings: assign.Findings{ + Put: func(k string, v []byte) error { held[k] = v; return nil }, + 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) + before := asked["ace"] + q.engineReads(w.nodes(nil, nil)) + clock = clock.Add(6 * time.Minute) + off, _ := presenceValue(false) + w.put(buckets.Presence, "ace", off) + q.engineReads(w.nodes(nil, nil)) + if len(held) != 0 { + t.Errorf("a finding was written while ace is away: %v", slices.Collect(maps.Keys(held))) + } + w.put(buckets.Presence, "ace", on) + q.engineReads(w.nodes(nil, nil)) + if _, ok := held[key]; !ok || asked["ace"]-before < 3 { + t.Errorf("ace unanswered six minutes, heard: findings %v, asked %d times", slices.Collect(maps.Keys(held)), asked["ace"]-before) + } + answers = true + q.engineReads(w.nodes(nil, nil)) + q.engineReads(w.nodes(nil, nil)) + if _, ok := held[key]; ok { + t.Errorf("the finding stays after ace answered") + } // The contract carries no checks, so its reader judges what the engine says (mesh-contracts seats/nodeengine): an // answer or an event that does not hold is no statement, and an event's node is its subject's last token. now := time.Now().Format(time.RFC3339Nano) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index 7c60c464..cf2b67f4 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -2,6 +2,7 @@ package main import ( "context" + "encoding/json" "errors" "maps" "mesh-delivery-assigner/internal/buckets" @@ -339,6 +340,24 @@ func presenceOf(key string, value []byte) (node string, online, ok bool) { return node, p.Online, true } +// seatHeard says a machine holding seat (a seat held once) is online as the watchman holds it: the callee of a call to +// the seat's verb is heard (novox/hq ADR 0668 §3). +func (w *world) seatHeard(seat string) bool { + online := w.online() + for _, value := range w.read(plc.SeatHoldersBucket) { + var sh mma.SeatHolders + if json.Unmarshal(value, &sh) != nil || sh.Seat != seat { + continue + } + for _, h := range sh.Holders { + if online[h.Node] { + return true + } + } + } + return false +} + // online is, by node, whether the watchman holds the node online. func (w *world) online() map[string]bool { out := map[string]bool{}