The checker moves a step that no build node answers to another node and fails naming every node tried (issue 1820)

modules/mesh-delivery-checker/cmd/mesh-delivery-checker/heard.go

@@ -0,0 +1,26 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+
+	"github.com/nats-io/nats.go"
+
+	"mesh-delivery-checker/internal/check"
+)
+
+// heardCaller is the build runner's caller with the calls that reached nobody marked: no responders, or none answering within
+// the call's bound. The step never began there, so the check may give it to another node (issue 1820). A call that was
+// answered, with a failure or not, is left as it is.
+type heardCaller struct {
+	call func(node, key string, body any) (json.RawMessage, error)
+}
+
+func (h heardCaller) CallOn(node, key string, body any) (json.RawMessage, error) {
+	raw, err := h.call(node, key, body)
+	if errors.Is(err, nats.ErrNoResponders) || errors.Is(err, nats.ErrTimeout) || errors.Is(err, context.DeadlineExceeded) {
+		return nil, &check.Unheard{Err: err}
+	}
+	return raw, err
+}

modules/mesh-delivery-checker/cmd/mesh-delivery-checker/heard_test.go

@@ -0,0 +1,30 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/nats-io/nats.go"
+
+	"mesh-delivery-checker/internal/check"
+)
+
+func TestACallThatReachedNobodyIsMarkedAndAnAnsweredFailureIsNot(t *testing.T) { // 1820/AC6
+	for _, err := range []error{nats.ErrNoResponders, nats.ErrTimeout, context.DeadlineExceeded} {
+		h := heardCaller{call: func(string, string, any) (json.RawMessage, error) { return nil, fmt.Errorf("call: %w", err) }}
+		_, got := h.CallOn("g14", "m.t", nil)
+		var u *check.Unheard
+		if !errors.As(got, &u) {
+			t.Errorf("%v: %v is not marked unheard", err, got)
+		}
+	}
+	h := heardCaller{call: func(string, string, any) (json.RawMessage, error) { return nil, errors.New("the module refused") }}
+	_, got := h.CallOn("g14", "m.t", nil)
+	var u *check.Unheard
+	if errors.As(got, &u) {
+		t.Errorf("an answered failure %v is marked unheard", got)
+	}
+}

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

@@ -189,7 +189,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	have := &buildrun.HaveSet{}
 	trunks := &check.Trunks{}
 	places := &buildrun.Places{Capacity: capacity}
-	runner := buildrun.Runner{Caller: caller, Steps: stepsWaiter{js}, Bound: stepBound}
+	runner := buildrun.Runner{Caller: heardCaller{caller.CallOn}, Steps: stepsWaiter{js}, Bound: stepBound}
 	ch := &check.Checker{
 		Verdicts: kvStore{buckets[delivery.CheckerVerdictsBucket]},
 		Owed:     kvStore{buckets[delivery.CheckerOwedBucket]},

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

@@ -0,0 +1,101 @@
+package check
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/nats-io/nats.go"
+)
+
+// Unheard is the error of a call that reached nobody: the bus had no responder for it, or none answered within the call's
+// bound. The step it asked for never began, so the same step may be asked of another node (issue 1820). The program wraps
+// the errors of its calls in it where the bus gives a timeout; a no-responders error is known as it is.
+type Unheard struct{ Err error }
+
+func (u *Unheard) Error() string { return u.Err.Error() }
+func (u *Unheard) Unwrap() error { return u.Err }
+
+// isUnheard says whether err is a call that reached nobody. A step that began and failed, or ended failed, is not.
+func isUnheard(err error) bool {
+	var u *Unheard
+	return errors.As(err, &u) || errors.Is(err, nats.ErrNoResponders)
+}
+
+// deafNode is a node that did not answer a build module's call during this judging.
+type deafNode struct{ node, why string }
+
+// markDeaf takes a node out of the choice for a build module for the rest of the judging.
+func (st *state) markDeaf(bm, node string, err error) {
+	if st.deaf == nil {
+		st.deaf = map[string][]deafNode{}
+	}
+	st.deaf[bm] = append(st.deaf[bm], deafNode{node, err.Error()})
+}
+
+func (st *state) isDeaf(bm, node string) bool {
+	for _, d := range st.deaf[bm] {
+		if d.node == node {
+			return true
+		}
+	}
+	return false
+}
+
+// triedWords names every node that did not answer a build module, with what the bus said.
+func (st *state) triedWords(bm string) string {
+	var parts []string
+	for _, d := range st.deaf[bm] {
+		parts = append(parts, fmt.Sprintf("%s (%s)", d.node, d.why))
+	}
+	return strings.Join(parts, "; ")
+}
+
+// moveStep is called when the target's node did not answer: the node is left out of the choice, and another build node of the
+// same build module takes the target's place. It answers an error naming every node tried when none is left (the words hold
+// NodesWords, so the verdict is judged again when the nodes change).
+func (st *state) moveStep(t *target, err error) error {
+	old := t.node
+	st.markDeaf(t.module, old, err)
+	node, _, cerr := st.choose(t.module)
+	if cerr != nil {
+		return fmt.Errorf("%s; tried: %s", cerr, st.triedWords(t.module))
+	}
+	st.j.logf("%s#%d: %s did not answer (%v), the step goes to %s", t.pull.Repository, t.pull.Number, old, err, node)
+	t.node = node
+	other := false
+	for _, o := range st.targets {
+		if o != t && o.node == old {
+			other = true
+		}
+	}
+	if !other {
+		delete(st.nodes, old)
+	}
+	st.nodes[node] = true
+	return nil
+}
+
+// refetch makes the target's workspace again on its new node: the workspace of a step that was lost is on the node that did
+// not answer.
+func (st *state) refetch(ctx context.Context, t *target) error {
+	raw, err := st.step(ctx, t, "fetch", fmt.Sprintf("fetch %s#%d on %s", t.pull.Repository, t.pull.Number, t.node), t.fetchArgs)
+	if err != nil {
+		return err
+	}
+	var f Fetched
+	if err := json.Unmarshal(raw, &f); err != nil || f.Workspace == "" {
+		return fmt.Errorf("the fetch of %s#%d on %s answered no workspace: %s", t.pull.Repository, t.pull.Number, t.node, raw)
+	}
+	t.fetched = f
+	return nil
+}
+
+// stepOnWorkspace sets the workspace a step reads to the target's current one.
+func stepOnWorkspace(t *target, args map[string]any) {
+	if _, ok := args["workspace"]; ok {
+		args["workspace"] = t.fetched.Workspace
+	}
+}

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

@@ -0,0 +1,142 @@
+package check
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/nats-io/nats.go"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// Issue 1820: a build step goes to a node whose build module is not heard (a laptop asleep, not yet marked offline by the
+// watchman), and the bus answers "no responders". The check moves the step to another node of the same build module.
+
+// deafNodes makes every tool on the named nodes fail as the bus does when nobody listens there; count says how many
+// times the node was called.
+func deafNodes(b *bench, deaf ...string) {
+	for tool, h := range b.steps.handlers {
+		h := h
+		b.steps.handlers[tool] = func(c stepCall) (any, error) {
+			for _, n := range deaf {
+				if c.node == n {
+					return nil, fmt.Errorf("%s on %s: %w", c.tool, c.node, nats.ErrNoResponders)
+				}
+			}
+			return h(c)
+		}
+	}
+}
+
+func nodesOf(calls []stepCall) []string {
+	var out []string
+	for _, c := range calls {
+		out = append(out, c.node)
+	}
+	return out
+}
+
+func TestAFetchThatNobodyAnswersMovesToAnotherBuildNode(t *testing.T) { // 1820/AC1
+	b := newBench(t)
+	b.ex.Toolchains = goEntries("ace", "g14")
+	deafNodes(b, "ace")
+	v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict
+	if v.State != delivery.VerdictPass {
+		t.Fatalf("verdict = %+v, want a pass on the node that answers", v)
+	}
+	if got := nodesOf(b.steps.tool("fetch")); len(got) != 2 || got[0] != "ace" || got[1] != "g14" {
+		t.Errorf("fetch went to %v, want ace (no responders) then g14", got)
+	}
+	for _, c := range b.steps.tool("test") {
+		if c.node != "g14" {
+			t.Errorf("a test went to %s, want g14 only", c.node)
+		}
+	}
+	if len(v.Machines) != 1 || v.Machines[0] != "g14" {
+		t.Errorf("machines = %v, want [g14]: ace ran nothing", v.Machines)
+	}
+}
+
+func TestATestStepThatNobodyAnswersIsFetchedAgainOnAnotherNode(t *testing.T) { // 1820/AC2
+	b := newBench(t)
+	b.ex.Toolchains = goEntries("ace", "g14")
+	// ace answers the fetch and the digest, then goes to sleep before the test.
+	test := b.steps.handlers["build_go_test"]
+	b.steps.handlers["build_go_test"] = func(c stepCall) (any, error) {
+		if c.node == "ace" {
+			return nil, fmt.Errorf("test on ace: %w", nats.ErrNoResponders)
+		}
+		return test(c)
+	}
+	v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict
+	if v.State != delivery.VerdictPass {
+		t.Fatalf("verdict = %+v, want a pass", v)
+	}
+	if got := nodesOf(b.steps.tool("fetch")); len(got) != 2 || got[1] != "g14" {
+		t.Errorf("fetch went to %v, want ace then g14 (the workspace was on ace)", got)
+	}
+	for _, c := range b.steps.tool("test") {
+		if c.node == "g14" && c.args["workspace"] == nil {
+			t.Errorf("a test on g14 has no workspace")
+		}
+	}
+	if len(v.Machines) != 1 || v.Machines[0] != "g14" {
+		t.Errorf("machines = %v, want [g14]", v.Machines)
+	}
+}
+
+func TestWhenNoBuildNodeAnswersTheFailureNamesEveryNodeTried(t *testing.T) { // 1820/AC3
+	b := newBench(t)
+	b.ex.Toolchains = goEntries("ace", "g14", "shanks")
+	deafNodes(b, "ace", "g14", "shanks")
+	v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict
+	if v.State != delivery.VerdictFail {
+		t.Fatalf("verdict = %+v, want a failure", v)
+	}
+	for _, n := range []string{"ace", "g14", "shanks"} {
+		if !strings.Contains(v.Failed.Words, n) {
+			t.Errorf("failure %q does not name %s", v.Failed.Words, n)
+		}
+	}
+	if !IsNodesFailure(v) {
+		t.Errorf("failure %q is not a failure of the nodes, so it would not be judged again when they change", v.Failed.Words)
+	}
+	if got := len(b.steps.tool("fetch")); got != 3 {
+		t.Errorf("fetch was tried %d times, want once on each of 3 nodes", got)
+	}
+}
+
+func TestAStepThatFailedForTheChangeIsNotMovedToAnotherNode(t *testing.T) { // 1820/AC4
+	b := newBench(t)
+	b.ex.Toolchains = goEntries("ace", "g14")
+	b.steps.handlers["build_go_fetch"] = func(stepCall) (any, error) { return nil, fmt.Errorf("fetch on ace: the step failed: git exploded") }
+	v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict
+	if v.State != delivery.VerdictFail || len(b.steps.tool("fetch")) != 1 {
+		t.Errorf("verdict = %+v after %d fetches, want a failure after one", v, len(b.steps.tool("fetch")))
+	}
+}
+
+func TestACanceledCheckIsNotMovedToAnotherNode(t *testing.T) { // 1820/AC4
+	b := newBench(t)
+	b.ex.Toolchains = goEntries("ace", "g14")
+	ctx, cancel := context.WithCancel(context.Background())
+	b.steps.handlers["build_go_fetch"] = func(stepCall) (any, error) {
+		cancel()
+		return nil, fmt.Errorf("fetch on ace: %w", nats.ErrNoResponders)
+	}
+	_, _ = b.ex.Judge(ctx, caseOf(t, pullOf(repoA, 7, head1)))
+	if n := len(b.steps.tool("fetch")); n != 1 {
+		t.Errorf("fetch was tried %d times after the check ended, want 1", n)
+	}
+}
+
+func TestAPinReadThatNobodyAnswersIsTheForgesFailureNotABuildNodes(t *testing.T) { // 1820/AC5
+	b := groupBench(t, "cb54756c3865")
+	b.ex.Forge.(*fakeForge).err = fmt.Errorf("gitea.gitea_get_file: %w", nats.ErrNoResponders)
+	v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1), pullOf(repoSDK, 12, head2))).Verdict
+	if v.State != delivery.VerdictFail || v.Failed.Machine != "forge" {
+		t.Errorf("verdict = %+v, want a failure at the forge: no build node was to blame", v)
+	}
+}

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

@@ -94,6 +94,8 @@ type target struct {
 	node      string
 	toolchain string
 	fetched   Fetched
+	// fetchArgs are the arguments of the target's fetch, so a workspace lost with its node is made again on another.
+	fetchArgs map[string]any
 	// priority are the Go packages (import paths) or the Node module directories whose test files the merge changes.
 	changed []string
 	// beside says the workspace was given siblings: its tests may read what the digest does not cover, so nothing is kept for it.
@@ -124,6 +126,7 @@ type state struct {
 	targets []*target
 	levels  map[string]*acc
 	nodes   map[string]bool
+	deaf    map[string][]deafNode // by build module: the nodes that did not answer during this judging
 	clone   time.Duration
 	slow    []string
 }
@@ -202,8 +205,28 @@ func fail(machine, words string) *delivery.Verdict {
 	return &v
 }
 
-// step runs one step of a build module on the node the target is on, holding a place there while it runs.
+// step runs one step of a build module on the node the target is on, holding a place there while it runs. A node that does
+// not answer (the bus has no responder, or none answered) is left out and the step goes to another node of the same build
+// module; a step after the fetch makes its workspace again there first (issue 1820).
 func (st *state) step(ctx context.Context, t *target, tool, label string, args map[string]any) (json.RawMessage, error) {
+	for {
+		raw, err := st.stepOnce(ctx, t, tool, label, args)
+		if err == nil || ctx.Err() != nil || !isUnheard(err) {
+			return raw, err
+		}
+		if merr := st.moveStep(t, err); merr != nil {
+			return nil, merr
+		}
+		if tool != "fetch" {
+			if rerr := st.refetch(ctx, t); rerr != nil {
+				return nil, rerr
+			}
+			stepOnWorkspace(t, args)
+		}
+	}
+}
+
+func (st *state) stepOnce(ctx context.Context, t *target, tool, label string, args map[string]any) (json.RawMessage, error) {
 	st.c.Progress.Set(label, t.node)
 	release, err := st.j.Places.Acquire(ctx, t.node)
 	if err != nil {
@@ -217,7 +240,7 @@ func (st *state) step(ctx context.Context, t *target, tool, label string, args m
 func (st *state) choose(bm string) (node, toolchain string, err error) {
 	language := buildrun.LanguageOf[bm]
 	qualifies := func(node string) bool {
-		return (st.j.Presence == nil || st.j.Presence.Online(node)) && (st.j.Health == nil || st.j.Health.Healthy(node, bm))
+		return !st.isDeaf(bm, node) && (st.j.Presence == nil || st.j.Presence.Online(node)) && (st.j.Health == nil || st.j.Health.Healthy(node, bm))
 	}
 	nodes, tc, err := buildrun.ChooseNodes(language, st.j.Toolchains.Entries(language), st.j.Places, qualifies, 1)
 	if err != nil {
@@ -241,6 +264,7 @@ func (st *state) fetchAll(ctx context.Context) *delivery.Verdict {
 			t := &target{pull: p, module: bm, node: node, toolchain: tc, plans: map[string]*goPlan{}}
 			st.nodes[node] = true
 			args := map[string]any{"repository": p.Repository, "commit": p.Head, "ref": refOf(p.Number), "base": p.Trunk}
+			t.fetchArgs = args
 			if bm == buildrun.GoBuildModule {
 				beside, err := st.j.besideFor(ctx, st.c, p)
 				if err != nil {
@@ -252,6 +276,7 @@ func (st *state) fetchAll(ctx context.Context) *delivery.Verdict {
 				}
 			}
 			raw, err := st.step(ctx, t, "fetch", fmt.Sprintf("fetch %s#%d on %s", p.Repository, p.Number, node), args)
+			node = t.node // the step may have moved to another node
 			if err != nil {
 				return fail(node, fmt.Sprintf("fetch of %s#%d: %v%s", p.Repository, p.Number, err, pinHint(st.c, err.Error())))
 			}

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

@@ -67,7 +67,7 @@ func (st *state) checkPins(ctx context.Context) verdictPtr {
 			st.c.Progress.Set(fmt.Sprintf("read %s of %s#%d", file, t.pull.Repository, t.pull.Number), t.node)
 			text, err := st.j.Forge.File(ctx, t.pull.Repository, file, t.pull.Head)
 			if err != nil {
-				return failed(t.node, fmt.Sprintf("%s#%d: %s cannot be read to check its mesh-sdk pin: %v", t.pull.Repository, t.pull.Number, file, err))
+				return failed(forgeMachine(err, t.node), fmt.Sprintf("%s#%d: %s cannot be read to check its mesh-sdk pin: %v", t.pull.Repository, t.pull.Number, file, err))
 			}
 			pin, ok := SDKPin(text)
 			if !ok {
@@ -78,7 +78,7 @@ func (st *state) checkPins(ctx context.Context) verdictPtr {
 			}
 			onTrunk, err := st.j.Forge.OnTrunk(ctx, SDKRepository, pin)
 			if err != nil {
-				return failed(t.node, fmt.Sprintf("%s#%d: whether mesh-sdk %s is on its trunk cannot be told: %v", t.pull.Repository, t.pull.Number, pin, err))
+				return failed(forgeMachine(err, t.node), fmt.Sprintf("%s#%d: whether mesh-sdk %s is on its trunk cannot be told: %v", t.pull.Repository, t.pull.Number, pin, err))
 			}
 			if onTrunk {
 				continue
@@ -104,3 +104,12 @@ func pinHint(c *Case, fetchError string) string {
 	}
 	return " (a pin to the head of the group's mesh-sdk pull request names a commit only that head carries, which go mod download cannot resolve; " + Issue1148 + ")"
 }
+
+// forgeMachine is the machine a failed read of the forge is put at: the forge when nobody answered it (no build node is to
+// blame, issue 1820), the build node whose workspace was read otherwise.
+func forgeMachine(err error, node string) string {
+	if isUnheard(err) {
+		return "forge"
+	}
+	return node
+}