D3: readers of whether an ask ended use Ended; the asker waits through a step given again and says a transport fault on N runs (ADR 0664 §3, issue 1748)

Shared lines for the join with milestone G: mesh-delivery-checker internal/check/read.go (one line), mesh-delivery-builder buildrun/asks.go.

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ

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

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
 	"time"
 
@@ -79,6 +80,7 @@ func (k Asker) Wait(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, e
 		}
 	}()
 	running, since, phase, phaseAt := "", time.Time{}, "", time.Time{}
+	var agains []string // the node of each run that ended on a transport fault, in order (ADR 0664 §3)
 	standStill := func() error {
 		if phase == "" {
 			return fmt.Errorf("the ask %s's step on %s did not end within %s of its start: it stood still, saying no phase", a.ID, running, k.Bound)
@@ -102,7 +104,24 @@ func (k Asker) Wait(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, e
 			if seen := stepSeenOf(ctx); seen != nil {
 				seen(step)
 			}
-			if step.State != buildstep.StepRunning {
+			if step.State == buildstep.StepAgain {
+				// A run ended on a transport fault and the bus gives the ask again: not an end. The bound runs on from the
+				// time it is given again, so a bus that never gives it says so (ADR 0664 §3).
+				agains = append(agains, step.Node)
+				running, since, phase, phaseAt = step.Node, step.Started, "given again after a transport fault", step.AgainAt
+				if k.Bound > 0 {
+					if timer != nil {
+						timer.Stop()
+					}
+					timer = time.NewTimer(max(time.Until(step.AgainAt.Add(k.Bound)), 0))
+					bound = timer.C
+				}
+				continue
+			}
+			if buildstep.Ended(step) {
+				if step.Passing && step.State == buildstep.StepFailed && len(agains) > 0 {
+					step.Reason = fmt.Sprintf("a transport fault on %d runs (%s): %s", len(agains)+1, strings.Join(append(agains, step.Node), ", "), step.Reason)
+				}
 				return step, nil
 			}
 			if k.Bound <= 0 {

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

@@ -40,6 +40,37 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
 	if _, err := k.Ask(context.Background(), goAsk("mvl0frv18d1958")); err == nil || !strings.Contains(err.Error(), "shanks refused: go 1.27 installed") {
 		t.Fatalf("a refused step: %v", err)
 	}
+	// Issue 1748 (ADR 0664 §3): a run that ended on a transport fault is no end; the next run's end is the answer, in either
+	// arrival order; three runs then failed say a transport fault on 3 runs with each node; a bus that never gives the ask
+	// again ends the wait saying so.
+	for name, order := range map[string][]buildstep.Step{
+		"again then passed": {{ID: "mvl0frv18d1971", Node: "ace", State: buildstep.StepAgain, Run: 1, AgainAt: time.Now()}, {ID: "mvl0frv18d1971", Node: "g14", State: buildstep.StepPassed, Run: 2}},
+		"passed then again": {{ID: "mvl0frv18d1971", Node: "g14", State: buildstep.StepPassed, Run: 2}, {ID: "mvl0frv18d1971", Node: "ace", State: buildstep.StepAgain, Run: 1, AgainAt: time.Now()}},
+	} {
+		h := &FakeAsks{}
+		for _, st := range order {
+			h.Write(st)
+		}
+		if got, err := (Asker{Publisher: h, Waiter: h, Bound: time.Minute}).Wait(context.Background(), goAsk("mvl0frv18d1971")); err != nil || got.State != buildstep.StepPassed {
+			t.Errorf("%s: %+v, %v", name, got, err)
+		}
+	}
+	h := &FakeAsks{}
+	for i, node := range []string{"ace", "g14", "shanks"} {
+		st := buildstep.Step{ID: "mvl0frv18d1972", Node: node, State: buildstep.StepAgain, Run: i + 1, AgainAt: time.Now()}
+		if i == 2 {
+			st.State, st.Passing, st.Reason = buildstep.StepFailed, true, "lookup proxy: i/o timeout"
+		}
+		h.Write(st)
+	}
+	if _, err := (Asker{Publisher: h, Waiter: h, Bound: time.Minute}).Ask(context.Background(), goAsk("mvl0frv18d1972")); err == nil || !strings.Contains(err.Error(), "a transport fault on 3 runs (ace, g14, shanks): lookup proxy: i/o timeout") {
+		t.Errorf("three runs then failed: %v", err)
+	}
+	h = &FakeAsks{}
+	h.Write(buildstep.Step{ID: "mvl0frv18d1973", Node: "ace", State: buildstep.StepAgain, Run: 1, AgainAt: time.Now(), Started: time.Now()})
+	if _, err := (Asker{Publisher: h, Waiter: h, Bound: 20 * time.Millisecond}).Wait(context.Background(), goAsk("mvl0frv18d1973")); err == nil || !strings.Contains(err.Error(), "given again after a transport fault") {
+		t.Errorf("an ask never given again: %v", err)
+	}
 	// review round 2 item 1: an ask that waits in its stream for a free slot is not bounded; once a node writes its step
 	// running, the step is bounded from its start, and the error names the node.
 	g := &FakeAsks{}

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

@@ -112,7 +112,7 @@ func (h *HeardEnds) WatchAsk(ctx context.Context, module, askID string) (<-chan
 					continue
 				}
 				for _, s := range steps {
-					if s.State != buildstep.StepRunning {
+					if buildstep.Ended(s) {
 						h.logf("the ask %s's step on %s ended %s, read at a nudge: its watch had not handed it over", askID, s.Node, s.State)
 					}
 					if !send(s) {
@@ -206,7 +206,7 @@ func (h *HeardEnds) follow(ctx context.Context, kv jetstream.KeyValue, module st
 				continue
 			}
 			heard = true
-			if step, err := buildstep.DecodeStep(e.Value()); err == nil && step.State != buildstep.StepRunning {
+			if step, err := buildstep.DecodeStep(e.Value()); err == nil && buildstep.Ended(step) {
 				h.Nudge(module)
 			}
 		}

modules/mesh-delivery-checker/internal/check/read.go

@@ -710,7 +710,7 @@ func (s stepsRead) lostWhy(c meshcatalog.CandidatesByGroupValue) string {
 	var last *buildstep.Step
 	for i := range s.all {
 		st := s.all[i].step
-		if st.State != buildstep.StepRunning && asksOf(c, st.What) && (last == nil || st.Ended.After(last.Ended)) {
+		if buildstep.Ended(st) && asksOf(c, st.What) && (last == nil || st.Ended.After(last.Ended)) {
 			last = &s.all[i].step
 		}
 	}

modules/mesh-delivery-checker/internal/check/try.go

@@ -231,7 +231,7 @@ func (t Try) HoldsANode() bool {
 	if len(t.Steps) == 0 {
 		return true
 	}
-	return slices.ContainsFunc(t.Steps, func(s TryStep) bool { return s.State == buildstep.StepRunning })
+	return slices.ContainsFunc(t.Steps, func(s TryStep) bool { return s.State == buildstep.StepRunning || s.State == buildstep.StepAgain })
 }
 
 // Over says whether the try came to its end.

modules/mesh-delivery-checker/internal/check/try_test.go

@@ -128,6 +128,10 @@ func TestATryIsAskedQueuedOncePerHeadAndAtMostFourAtOnce(t *testing.T) { // ADR
 			t.Errorf("%s: answered %v, want refused %q", c.what, err, c.refused)
 		}
 	}
+	// Issue 1748: a try whose step ended on a transport fault, the bus giving its ask again, still runs.
+	if !(Try{State: TryRunning, Steps: []TryStep{{Ask: "a", State: buildstep.StepAgain}}}).HoldsANode() {
+		t.Error("a try with a step given again no longer runs")
+	}
 }
 
 func TestATryOfARepositoryWithNoCheckedLevelIsRefused(t *testing.T) { // the agreed error of the try

sdk/buildrun/stale_test.go

@@ -83,6 +83,7 @@ func TestEndStaleAndEndOne(t *testing.T) {
 	put("ace", "raced1", "build_go_test", buildstep.StepRunning, boot.Add(-time.Hour), "")
 	put("ace", "new1", "build_go_fetch", buildstep.StepRunning, boot.Add(time.Second), "")
 	put("ace", "done1", "build_go_fetch", buildstep.StepPassed, boot.Add(-time.Hour), gone)
+	put("ace", "again1", "build_go_fetch", buildstep.StepAgain, boot.Add(-time.Hour), gone) // issue 1748: the bus gives it again; the sweep leaves it
 	put("g14", "other1", "build_go_fetch", buildstep.StepRunning, boot.Add(-time.Hour), gone)
 	// raced1 is ended by its own run between the read and the write.
 	raced := false
@@ -97,6 +98,9 @@ func TestEndStaleAndEndOne(t *testing.T) {
 	if err != nil || len(ended) != 4 {
 		t.Fatalf("ended %v, %v", ended, err)
 	}
+	if s := k.last("ace.again1"); s.State != buildstep.StepAgain {
+		t.Errorf("the sweep ended a step the bus gives again: %+v", s)
+	}
 	if _, held := k.puts["ace.mjk3l9x0abcded"]; held {
 		t.Errorf("the ghost ask's key is still held: %+v", k.last("ace.mjk3l9x0abcded"))
 	}