D9: the assigner keeps the wait of its bind's asks read and of a heard node's manifest-reads, and says each unanswered past five minutes as call-unanswered (issue 1966, ADR 0668 §3; review round 1, finding 4)

The bind's read of the operator channel's asks is made again after every transport fault until answered; a node whose engine answers nothing while its machine is heard is asked again at each pass once its ask ended. Past CallUnansweredAfter with the callee's machine heard each writes the finding in the assigner's findings, cleared once answered; none while the machine is away. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
2026-11-10 04:17:45 +01:00
parent 8f50f22622
commit 23ddaa8bed
5 changed files with 217 additions and 6 deletions
@@ -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
 	}
@@ -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 {
@@ -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 {
@@ -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)
@@ -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{}