BL17e: the read of an ask's steps is a kept call: a fault ends no key, the watch is taken up again from its last revision, past five minutes unanswered it is call-unanswered on a bound key

JetStreamAsks.WatchAsk opens the watch as a kept call and takes a closed watch up after the last revision heard; the first
fault arms a key of the state bounds with its own time to live (StepsBoundKey), whose end while the read is unanswered raises
call-unanswered about the ask, and the answer ends it.

Builds: design 47 G §568, A-77
Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h

modules/mesh-delivery-builder/buildrun/asks.go

@@ -299,6 +299,14 @@ func (k Asker) Passed(step buildstep.Step, err error) (buildstep.Step, error) {
 // of `*.<ask id>` in the build module's steps bucket.
 type JetStreamAsks struct {
 	JS jetstream.JetStream
+	// Unanswered is told each time the read of an ask's steps meets a transport fault it keeps waiting out (design 47 G §568, ADR
+	// 0668 §3): the builder arms the call's bound, a key with its own time to live (A-77), and raises call-unanswered when it
+	// ends unanswered. Answered is told when the read is answered again. Both may be nil.
+	Unanswered func(module, askID string, fault error)
+	Answered   func(module, askID string)
+	// OpenWatch opens the watch of keys filtered by pattern in the build module's steps state, from the revision after `after`
+	// when after is not 0; nil is the bus's.
+	OpenWatch func(ctx context.Context, module, pattern string, after uint64) (jetstream.KeyWatcher, error)
 }
 
 // PublishAsk is AskPublisher's.
@@ -309,46 +317,100 @@ func (s JetStreamAsks) PublishAsk(ctx context.Context, subject, id string, data
 	return err
 }
 
-// WatchAsk is AskWaiter's: the watch hands over what the keys hold now, so a step that ended before it began is read at
-// once, and every change after.
-func (s JetStreamAsks) WatchAsk(ctx context.Context, module, askID string) (<-chan buildstep.Step, error) {
-	pattern, err := buildstep.AskStepWatch(askID)
-	if err != nil {
-		return nil, err
+// openWatch opens the watch of pattern in the steps state of module, from the revision after `after` when it is not 0.
+func (s JetStreamAsks) openWatch(ctx context.Context, module, pattern string, after uint64) (jetstream.KeyWatcher, error) {
+	if s.OpenWatch != nil {
+		return s.OpenWatch(ctx, module, pattern, after)
 	}
 	kv, err := s.JS.KeyValue(ctx, StepsBucket(module))
 	if err != nil {
 		return nil, fmt.Errorf("the bucket %s: %w", StepsBucket(module), err)
 	}
-	w, err := kv.Watch(ctx, pattern)
+	if after > 0 {
+		return kv.Watch(ctx, pattern, jetstream.ResumeFromRevision(after))
+	}
+	return kv.Watch(ctx, pattern)
+}
+
+// transportFault says err is a transport fault of ADR 0664 §1's classes or the read's own deadline: nothing was answered.
+func transportFault(err error) bool {
+	return busretry.Undelivered(err) || busretry.TimedOut(err)
+}
+
+// openKept is a kept call: the watch is opened, and opened again while the bus answers nothing, until it is or ctx ends
+// (design 47 G §568: a transport fault of the builder's own read ends no build key). Each fault past the call's own asking again
+// is told to Unanswered; the answer, to Answered. An error that is no transport fault is the caller's.
+func (s JetStreamAsks) openKept(ctx context.Context, module, askID, pattern string, after uint64) (jetstream.KeyWatcher, error) {
+	for {
+		w, err := busretry.DoRepeatable("opening the watch of the ask "+askID+"'s step on "+module, func() (jetstream.KeyWatcher, error) {
+			return s.openWatch(ctx, module, pattern, after)
+		}, nil)
+		if err == nil {
+			if s.Answered != nil {
+				s.Answered(module, askID)
+			}
+			return w, nil
+		}
+		if ctx.Err() != nil {
+			return nil, ctx.Err()
+		}
+		if !transportFault(err) {
+			return nil, err
+		}
+		if s.Unanswered != nil {
+			s.Unanswered(module, askID, err)
+		}
+	}
+}
+
+// WatchAsk is AskWaiter's: the watch hands over what the keys hold now, so a step that ended before it began is read at
+// once, and every change after. A watch that cannot be opened for a transport fault is opened again once the bus answers, and one
+// that closes is taken up again from the revision after the last it handed over (design 47 G §568), so a fault fails no key.
+func (s JetStreamAsks) WatchAsk(ctx context.Context, module, askID string) (<-chan buildstep.Step, error) {
+	pattern, err := buildstep.AskStepWatch(askID)
+	if err != nil {
+		return nil, err
+	}
+	w, err := s.openKept(ctx, module, askID, pattern, 0)
 	if err != nil {
 		return nil, err
 	}
 	out := make(chan buildstep.Step)
 	go func() {
 		defer close(out)
-		defer w.Stop()
+		var last uint64
 		for {
-			select {
-			case <-ctx.Done():
-				return
-			case e, ok := <-w.Updates():
-				if !ok {
-					return
-				}
-				if e == nil || e.Operation() != jetstream.KeyValuePut {
-					continue
-				}
-				step, err := buildstep.DecodeStep(e.Value())
-				if err != nil {
-					continue // not a step (a node's pause key never matches an ask id; anything else is not this ask's)
-				}
+			closed := false
+			for !closed {
 				select {
-				case out <- step:
 				case <-ctx.Done():
+					_ = w.Stop()
 					return
+				case e, ok := <-w.Updates():
+					if !ok {
+						closed = true
+						break
+					}
+					if e == nil || e.Operation() != jetstream.KeyValuePut {
+						continue
+					}
+					last = max(last, e.Revision())
+					step, err := buildstep.DecodeStep(e.Value())
+					if err != nil {
+						continue // not a step (a node's pause key never matches an ask id; anything else is not this ask's)
+					}
+					select {
+					case out <- step:
+					case <-ctx.Done():
+						_ = w.Stop()
+						return
+					}
 				}
 			}
+			_ = w.Stop()
+			if w, err = s.openKept(ctx, module, askID, pattern, last+1); err != nil {
+				return
+			}
 		}
 	}()
 	return out, nil

modules/mesh-delivery-builder/buildrun/steps_resume_test.go

@@ -0,0 +1,88 @@
+package buildrun
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-delivery/sdk/buildstep"
+	"git.novox.be/novox/mesh-sdk/go/busretry"
+)
+
+type fakeEntry struct {
+	jetstream.KeyValueEntry
+	rev   uint64
+	value []byte
+}
+
+func (e fakeEntry) Revision() uint64                { return e.rev }
+func (e fakeEntry) Value() []byte                   { return e.value }
+func (e fakeEntry) Operation() jetstream.KeyValueOp { return jetstream.KeyValuePut }
+
+type fakeWatcher struct {
+	jetstream.KeyWatcher
+	updates chan jetstream.KeyValueEntry
+}
+
+func (w fakeWatcher) Updates() <-chan jetstream.KeyValueEntry { return w.updates }
+func (w fakeWatcher) Stop() error                             { return nil }
+
+// Design 47 G §568: the read of an ask's steps is a kept call. A transport fault opening the watch ends no key (the wait is
+// not failed): it is opened again once the bus answers, said unanswered meanwhile and answered after; and a watch that closes
+// is taken up again from the revision after the last it handed over, so the step that ended meanwhile is heard.
+func TestTheWatchOfAStepResumesAfterAFault(t *testing.T) {
+	t.Cleanup(busretry.Waits(20*time.Millisecond, time.Millisecond, 2*time.Millisecond))
+	passed, _ := buildstep.EncodeStep(buildstep.Step{ID: "mvl0frv18d1958", Tool: "build", Node: "ace", State: buildstep.StepPassed})
+	running, _ := buildstep.EncodeStep(buildstep.Step{ID: "mvl0frv18d1958", Tool: "build", Node: "ace", State: buildstep.StepRunning})
+	var mu sync.Mutex
+	var opens []uint64
+	var unanswered, answered int
+	asks := JetStreamAsks{
+		Unanswered: func(string, string, error) { mu.Lock(); unanswered++; mu.Unlock() },
+		Answered:   func(string, string) { mu.Lock(); answered++; mu.Unlock() },
+		OpenWatch: func(_ context.Context, _, _ string, after uint64) (jetstream.KeyWatcher, error) {
+			mu.Lock()
+			defer mu.Unlock()
+			opens = append(opens, after)
+			ch := make(chan jetstream.KeyValueEntry, 2)
+			// The bus answers nothing past the kept call's own asking again (its bound), so the fault is told and the watch tried again.
+			switch n := len(opens); {
+			case n <= 40:
+				return nil, nats.ErrNoResponders
+			case n == 41:
+				ch <- fakeEntry{rev: 5, value: running}
+				close(ch) // the watch closes under the wait
+			default:
+				ch <- fakeEntry{rev: 6, value: passed}
+			}
+			return fakeWatcher{updates: ch}, nil
+		}}
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	steps, err := asks.WatchAsk(ctx, "mesh-build-go", "mvl0frv18d1958")
+	if err != nil {
+		t.Fatalf("a transport fault opening the watch failed the wait: %v", err)
+	}
+	var states []string
+	for s := range steps {
+		states = append(states, s.State)
+		if buildstep.Ended(s) {
+			break
+		}
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	if len(states) != 2 || states[0] != buildstep.StepRunning || states[1] != buildstep.StepPassed {
+		t.Fatalf("steps heard = %v, want running then passed", states)
+	}
+	if len(opens) != 42 || opens[41] != 6 {
+		t.Errorf("opens after revisions %v, want the last taken up again after the last revision heard (6)", opens)
+	}
+	if unanswered == 0 || answered != 2 {
+		t.Errorf("unanswered %d, answered %d, want the fault said and each answer said", unanswered, answered)
+	}
+}

modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go

@@ -140,7 +140,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	tools := &buildrun.ToolchainSet{}
 	// Each build module's part of a build is an ask on its build seat's stream, pulled by a build node with a free slot
 	// (novox/hq ADR 0525): the builder chooses no node, so it reads neither presence nor have and holds no capacity.
-	asks := buildrun.JetStreamAsks{JS: js}
+	// The read of an ask's steps is a kept call (design 47 G §568): a fault ends no key, and past five minutes unanswered it is a
+	// finding; its bound is a key of the state bounds, bound once the bus holds it (set below, with the findings).
+	stepReads := &stepsRead{bounds: func() jetstream.KeyValue { return nil }, bound: meshconditionsclient.CallUnansweredAfter, now: time.Now, logf: logger.Printf}
+	asks := buildrun.JetStreamAsks{JS: js, Unanswered: stepReads.unanswered, Answered: stepReads.answered}
 	// The asks the build seats gave up on (issue 734): heard from DEAD_LETTERS, each ends the wait for its step at once,
 	// and the lane asks its part again (Builder.waitOnce) instead of waiting for the check bound.
 	// Heard in the background, and again at a reconnect or an ask while it is not; whether it is heard is kept in the
@@ -181,6 +184,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// A publish that ends badly is a finding (design 51 *Health and conditions*): a holder's refused, a version no holder
 	// answered within its bound (a key of the state bounds with its own time to live), a seat's outcome that does not read.
 	pubBoundsKV := process.AwaitBucket(ctx, js, conn, versions.BoundsBucket, "a publish no holder answers, and a node holding the old form seven days, are not raised", logger.Printf)
+	stepReads.bounds, stepReads.findings = func() jetstream.KeyValue { return pubBoundsKV.KV() }, found
 	pubs := &publishes{bounds: pubBoundsKV.KV, versions: numberer.Versions, findings: found, bound: versions.PublishBound, logf: logger.Printf}
 	numberer.Publishing = pubs.arm
 	b := &build.Builder{
@@ -443,6 +447,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 				unput.expired(ctx, e)
 				return
 			}
+			if _, _, ok := versions.StepsBoundOf(e.Key()); ok {
+				stepReads.expired(ctx, e)
+				return
+			}
 			if node, ok := enginereads.BoundOf(e.Key()); ok {
 				if e.Operation() == jetstream.KeyValuePurge { // its five minutes ended
 					reads.expiredCall(node)

modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepsread.go

@@ -0,0 +1,95 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
+
+	"mesh-delivery-builder/internal/versions"
+)
+
+// **The builder's read of an ask's steps is a kept call, unanswered past five minutes a finding** (design 47 G §568, A-77, ADR
+// 0668 §3): a transport fault opening or taking up the watch of `steps` ends no build key; the read is made again once the bus
+// answers. The first fault arms a key of the state bounds (StepsBoundKey) with its own time to live; the bus's marker of its end,
+// heard on the watch of the state, raises call-unanswered about the ask while the read is still unanswered; the answer deletes
+// the key and clears the finding. No timer.
+type stepsRead struct {
+	bounds   func() jetstream.KeyValue
+	findings *findings
+	bound    time.Duration
+	now      func() time.Time
+	logf     func(string, ...any)
+
+	mu   sync.Mutex
+	open map[string]string // the asks whose read is unanswered, by ask id: the last fault
+}
+
+// unanswered is told each fault of the read of the ask's steps on module: the first arms the bound.
+func (s *stepsRead) unanswered(module, askID string, fault error) {
+	s.mu.Lock()
+	if s.open == nil {
+		s.open = map[string]string{}
+	}
+	_, armed := s.open[askID]
+	s.open[askID] = fault.Error()
+	s.mu.Unlock()
+	if armed {
+		return
+	}
+	ctx := context.Background()
+	var err error
+	if kv := s.bounds(); kv == nil {
+		err = fmt.Errorf("%s is not on the bus yet", versions.BoundsBucket)
+	} else if _, err = kv.Create(ctx, versions.StepsBoundKey(module, askID), versions.EncodeBound("the read of the steps of the ask "+askID+" on "+module, s.now().Add(s.bound)), jetstream.KeyTTL(s.bound)); errors.Is(err, jetstream.ErrKeyExists) {
+		err = nil
+	}
+	if err != nil {
+		s.logf("the bound of the read of the steps of the ask %s is not written, so it is raised now: %v", askID, err)
+		s.raise(ctx, module, askID, fault.Error(), s.now())
+	}
+}
+
+// answered ends the read's bound and clears its finding.
+func (s *stepsRead) answered(module, askID string) {
+	s.mu.Lock()
+	_, was := s.open[askID]
+	delete(s.open, askID)
+	s.mu.Unlock()
+	if !was {
+		return
+	}
+	ctx := context.Background()
+	if kv := s.bounds(); kv != nil {
+		if err := kv.Delete(ctx, versions.StepsBoundKey(module, askID)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
+			s.logf("the bound of the read of the steps of the ask %s is not deleted: %v", askID, err)
+		}
+	}
+	s.findings.clear(ctx, meshconditionsclient.KindCallUnanswered, meshconditionsclient.ScopeCall, askID)
+}
+
+// expired is told each change of the state bounds: the end of a steps bound raises call-unanswered while its read is unanswered.
+func (s *stepsRead) expired(ctx context.Context, e jetstream.KeyValueEntry) {
+	module, askID, ok := versions.StepsBoundOf(e.Key())
+	if !ok || e.Operation() != jetstream.KeyValuePurge {
+		return
+	}
+	s.mu.Lock()
+	fault, open := s.open[askID]
+	s.mu.Unlock()
+	if !open {
+		return
+	}
+	s.raise(ctx, module, askID, fault, s.now().Add(-s.bound))
+}
+
+func (s *stepsRead) raise(ctx context.Context, module, askID, fault string, since time.Time) {
+	o := callUnanswered(askID, "the read of the steps of "+module, "", fault, "builder."+meshconditionsclient.KindCallUnanswered, since)
+	o.needs = "the bus answers the read again by itself; builder_stuck shows the key waiting on the ask."
+	s.findings.raise(ctx, o)
+}

modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepsread_test.go

@@ -0,0 +1,43 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
+
+	"mesh-delivery-builder/internal/versions"
+)
+
+// Design 47 G §568, A-77: the first fault of the read of an ask's steps arms a bound key with its own time to live; its end
+// while the read is unanswered raises call-unanswered about the ask; the answer deletes the bound and clears the finding.
+func TestTheStepsReadIsCallUnansweredAfterItsBound(t *testing.T) {
+	st := &memFindings{vals: map[string][]byte{}}
+	f := &findings{w: st, logf: t.Logf, now: func() time.Time { return time.Unix(0, 0) }}
+	marks := &memBounds{keys: map[string]bool{}}
+	s := &stepsRead{bounds: func() jetstream.KeyValue { return marks }, findings: f, bound: meshconditionsclient.CallUnansweredAfter,
+		now: func() time.Time { return time.Unix(600, 0) }, logf: t.Logf}
+	ask := "mvl0frv18d1958"
+	callKey := meshconditionsclient.Key(meshconditionsclient.ScopeCall, ask, meshconditionsclient.KindCallUnanswered)
+	s.unanswered("mesh-build-go", ask, errors.New("nats: no responders available for request"))
+	if !marks.keys[versions.StepsBoundKey("mesh-build-go", ask)] || st.vals[callKey] != nil {
+		t.Fatalf("a first fault: bound %v, finding %v; want the bound armed, no finding yet", marks.keys, st.vals[callKey] != nil)
+	}
+	s.expired(context.Background(), purged{key: versions.StepsBoundKey("mesh-build-go", ask)})
+	if got, err := decodeFinding(st.vals[callKey]); err != nil || got.Kind != meshconditionsclient.KindCallUnanswered || !strings.Contains(got.Summary, "the steps of mesh-build-go") || !strings.Contains(got.Summary, ask) {
+		t.Fatalf("the bound's end: the finding = %+v, %v", got, err)
+	}
+	s.answered("mesh-build-go", ask)
+	if marks.keys[versions.StepsBoundKey("mesh-build-go", ask)] || st.vals[callKey] != nil {
+		t.Errorf("the read answered: bound %v, finding left %v", marks.keys, st.vals[callKey] != nil)
+	}
+	s.expired(context.Background(), purged{key: versions.StepsBoundKey("mesh-build-go", ask)})
+	if st.vals[callKey] != nil {
+		t.Error("a bound's end after the answer raised call-unanswered")
+	}
+}

modules/mesh-delivery-builder/internal/versions/publish.go

@@ -37,6 +37,7 @@ const (
 	publishBoundPrefix = "publish."
 	oldFormBoundPrefix = "old-form."
 	unputBoundPrefix   = "unput."
+	stepsBoundPrefix   = "steps."
 )
 
 // UnputBoundKey is the bound of a build key a look left building because its end could not be written (issue 1966,
@@ -44,6 +45,21 @@ const (
 // finding call-unanswered about the key.
 func UnputBoundKey(buildKey string) string { return unputBoundPrefix + buildKey }
 
+// StepsBoundKey is the bound of the builder's read of an ask's steps while the bus does not answer it (design 47 G §568, A-77):
+// `steps.<build module>.<ask id>`. Its end, the bus's marker of its time to live, while the read is unanswered, is the finding
+// call-unanswered about the ask.
+func StepsBoundKey(module, askID string) string { return stepsBoundPrefix + module + "." + askID }
+
+// StepsBoundOf is the build module and ask a steps bound names, and whether key is one.
+func StepsBoundOf(key string) (module, askID string, ok bool) {
+	rest, ok := strings.CutPrefix(key, stepsBoundPrefix)
+	if !ok {
+		return "", "", false
+	}
+	module, askID, ok = strings.Cut(rest, ".")
+	return module, askID, ok && module != "" && askID != ""
+}
+
 // UnputBoundOf is the build key an unput bound names, and whether key is one.
 func UnputBoundOf(key string) (string, bool) { return strings.CutPrefix(key, unputBoundPrefix) }