diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/heard.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/heard.go new file mode 100644 index 0000000..e186fce --- /dev/null +++ b/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 +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/heard_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/heard_test.go new file mode 100644 index 0000000..d660385 --- /dev/null +++ b/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) + } +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index 752a6ba..40ed0d7 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/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]}, diff --git a/modules/mesh-delivery-checker/internal/check/heard.go b/modules/mesh-delivery-checker/internal/check/heard.go new file mode 100644 index 0000000..d97c17f --- /dev/null +++ b/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 + } +} diff --git a/modules/mesh-delivery-checker/internal/check/heard_test.go b/modules/mesh-delivery-checker/internal/check/heard_test.go new file mode 100644 index 0000000..784ffc7 --- /dev/null +++ b/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) + } +} diff --git a/modules/mesh-delivery-checker/internal/check/judge.go b/modules/mesh-delivery-checker/internal/check/judge.go index 8cdeb58..df003a4 100644 --- a/modules/mesh-delivery-checker/internal/check/judge.go +++ b/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()))) } diff --git a/modules/mesh-delivery-checker/internal/check/pins.go b/modules/mesh-delivery-checker/internal/check/pins.go index 7ce5df6..cbedcaa 100644 --- a/modules/mesh-delivery-checker/internal/check/pins.go +++ b/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 +}