diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go index e2f4c15..502c82f 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go @@ -5,6 +5,7 @@ package main import ( "context" "encoding/json" + "errors" "fmt" "io" "os" @@ -20,6 +21,7 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/announce" + "git.novox.be/novox/mesh-sdk/go/buildstep" "git.novox.be/novox/mesh-sdk/go/delivery" "mesh-delivery-checker/internal/check" @@ -75,7 +77,6 @@ func readCaptured(t *testing.T, name string) map[string]any { type fakeBuildModules struct { js jetstream.JetStream ctx context.Context - n atomic.Int64 conflict atomic.Bool mu sync.Mutex calls map[string]int // by `.` @@ -112,45 +113,55 @@ func (f *fakeBuildModules) result(t *testing.T, tool string, args map[string]any return data } -func (f *fakeBuildModules) serve(t *testing.T, nc *nats.Conn, module string) { - steps, err := f.js.CreateKeyValue(f.ctx, jetstream.KeyValueConfig{Bucket: delivery.StepsBucket(module), History: 1}) +// serve makes module's steps bucket and its build seat's stream and workers, and pulls its asks on n1 as the real module +// does since R4g (ADR 0525; issue 1887 A5): each ask's job runs with step bodies answering the captured results, and a +// check takes the kept verdicts of the checker's kept bucket by their digests. +func (f *fakeBuildModules) serve(t *testing.T, module string, kept jetstream.KeyValue) { + kv, err := f.js.CreateKeyValue(f.ctx, jetstream.KeyValueConfig{Bucket: delivery.StepsBucket(module), History: 1}) if err != nil { t.Fatal(err) } - sub, err := nc.Subscribe("mesh.mod."+module+".call.>", func(m *nats.Msg) { - tokens := strings.Split(m.Subject, ".") // mesh mod call - tool, node := tokens[4], tokens[5] - f.mu.Lock() - f.calls[module+"."+tool]++ - f.mu.Unlock() - var args map[string]any - _ = json.Unmarshal(m.Data, &args) - id := fmt.Sprintf("s%d", f.n.Add(1)) - key, _ := delivery.StepKey(node, id) - put := func(s delivery.Step) { - data, err := delivery.EncodeStep(s) - if err == nil { - _, err = steps.Put(f.ctx, key, data) + language := strings.TrimPrefix(module, "mesh-build-") + if err := buildstep.DeclareAskWorkers(f.ctx, f.js, language, 30*time.Second); err != nil { + t.Fatal(err) + } + prefix := map[string]string{"mesh-build-go": "build_go_", "mesh-build-node": "build_node_"}[module] + call := func(name string) buildstep.StepFunc { + return func(ctx context.Context, args map[string]any) (any, error) { + tool := prefix + name + f.mu.Lock() + f.calls[module+"."+tool]++ + f.mu.Unlock() + return f.result(t, tool, args), nil + } + } + steps := buildstep.AskSteps{Fetch: call("fetch"), Test: call("test"), + Kept: func(ctx context.Context, level, digest string) (int, bool, error) { + key, err := delivery.KeptKey(level, digest) + if err != nil { + return 0, false, err + } + e, err := kept.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return 0, false, nil } if err != nil { - t.Errorf("writing step %s: %v", key, err) + return 0, false, err } - } - step := delivery.Step{ID: id, Tool: tool, Node: node, State: delivery.StepRunning, Started: time.Now().UTC()} - put(step) - answer, _ := json.Marshal(map[string]any{"result": map[string]any{"step": id, "key": key, "state": "running"}, "error": "", "node": node}) - _ = m.Respond(answer) - go func() { - time.Sleep(20 * time.Millisecond) - step.Ended = time.Now().UTC() - step.State, step.Result = delivery.StepPassed, f.result(t, tool, args) - put(step) - }() - }) - if err != nil { - t.Fatal(err) + v, err := delivery.DecodeKept(e.Value()) + return v.Tests, err == nil, err + }} + if module == "mesh-build-go" { + steps.Digest = call("digest") } - t.Cleanup(func() { _ = sub.Unsubscribe() }) + put := func(key string, value any) (uint64, error) { + raw, _ := value.(json.RawMessage) + return kv.Put(f.ctx, key, raw) + } + p := &buildstep.Puller{Source: &buildstep.JetStreamAsks{JS: f.js, Language: language}, Language: language, Node: "n1", Put: put, + Capacity: func() int { return 2 }, PullWait: 200 * time.Millisecond, + Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return buildstep.RunAsk(ctx, steps, a) }} + go func() { _ = p.Pull(f.ctx) }() } // serveForge answers the forge's two reads as its tools do, with the answers the forge gave (internal/check/testdata/captured): @@ -283,8 +294,8 @@ func startBed(t *testing.T) *bed { } b.serveForge(t, ctl) b.fake = &fakeBuildModules{js: js, ctx: ctx, calls: map[string]int{}} - b.fake.serve(t, ctl, "mesh-build-go") - b.fake.serve(t, ctl, "mesh-build-node") + b.fake.serve(t, "mesh-build-go", b.buckets[delivery.CheckerKeptBucket]) + b.fake.serve(t, "mesh-build-node", b.buckets[delivery.CheckerKeptBucket]) b.report(t, "mesh-build-go", "n1", "go", "go1.27.1", "1") b.report(t, "mesh-build-node", "n1", "typescript", "tsc5", "1") b.setTrunk(t, map[string]string{repo: trunk}) diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go index 7db6c2b..8ea203b 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go @@ -15,11 +15,8 @@ import ( // judgeDeps are the parts the judge of a check is made of. type judgeDeps struct { - runner buildrun.Runner + asks check.Asks tools *buildrun.ToolchainSet - presence *buildrun.PresenceSet - have *buildrun.HaveSet - places *buildrun.Places trunks *check.Trunks kept check.KV previews check.KV @@ -31,7 +28,7 @@ type judgeDeps struct { } func newJudge(d judgeDeps) check.Judge { - return &check.Examiner{Steps: d.runner, Toolchains: d.tools, Presence: d.presence, Health: d.have, Places: d.places, + return &check.Examiner{Asks: d.asks, Toolchains: d.tools, Kept: check.Kept{Store: d.kept}, Previews: d.previews, Modules: d.repos, Packages: d.repos, Forge: d.forge, Runs: d.repos, Builds: d.builds, Announce: d.announce, Trunks: d.trunks, Logf: d.logf} } 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 41377e8..fb98e28 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -201,8 +201,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro presence := &buildrun.PresenceSet{} have := &buildrun.HaveSet{} trunks := &check.Trunks{} - places := &buildrun.Places{Capacity: capacity} - runner := buildrun.Runner{Caller: heardCaller{caller.CallOn}, Steps: stepsWaiter{js}, Bound: stepBound} + _ = capacity + // A check is one ask per pull request and build module on the build seats' check stream, pulled by a build node with a + // free slot (novox/hq ADR 0525 §1, §3): the checker chooses no node and starts no step. + asks := buildrun.JetStreamAsks{JS: js} ch := &check.Checker{ Verdicts: kvStore{buckets[delivery.CheckerVerdictsBucket]}, Owed: kvStore{buckets[delivery.CheckerOwedBucket]}, @@ -235,7 +237,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } repos := &repoModules{manifests: kvStore{buckets[delivery.ManifestsBucket]}, logf: logger.Printf} ch.Judge = newJudge(judgeDeps{ - runner: runner, tools: tools, presence: presence, have: have, places: places, trunks: trunks, + asks: buildrun.Asker{Publisher: asks, Waiter: asks, Bound: stepBound}, tools: tools, trunks: trunks, kept: kvStore{buckets[delivery.CheckerKeptBucket]}, previews: kvStore{buckets[delivery.PreviewsBucket]}, repos: repos, builds: builds, forge: forgeReads{call: caller.Call}, logf: logger.Printf, announce: announce, diff --git a/modules/mesh-delivery-checker/internal/check/case.go b/modules/mesh-delivery-checker/internal/check/case.go index 4240ed4..e7e8654 100644 --- a/modules/mesh-delivery-checker/internal/check/case.go +++ b/modules/mesh-delivery-checker/internal/check/case.go @@ -49,6 +49,9 @@ type Case struct { Try *TryOptions // Progress says what the check is doing, so a check that ran past its bound names the step it was stuck at. Progress *Progress + // Kind is what asked the check (KindNew, KindNodes, KindTrunk): a re-judge after a trunk move is asked on the build + // seats' recheck stream, every other check on check (ADR 0536 §4). + Kind string } // Progress is the step a check is at, written by the check and read when its bound passes. diff --git a/modules/mesh-delivery-checker/internal/check/checker.go b/modules/mesh-delivery-checker/internal/check/checker.go index 33f795e..850b7fe 100644 --- a/modules/mesh-delivery-checker/internal/check/checker.go +++ b/modules/mesh-delivery-checker/internal/check/checker.go @@ -207,7 +207,8 @@ func (ch *Checker) consider(ctx context.Context, key string, c delivery.Candidat go func() { defer ch.wg.Done() for { - ch.admit(cs, ch.kindOf(ctx, cs)) + cs.Kind = ch.kindOf(ctx, cs) // a re-judge after a trunk move is asked on the recheck stream (ADR 0536 §4) + ch.admit(cs, cs.Kind) ch.try(ctx, key, cs) ch.leave() ch.mu.Lock() diff --git a/modules/mesh-delivery-checker/internal/check/examiner_test.go b/modules/mesh-delivery-checker/internal/check/examiner_test.go index 88ce8fb..b90ae93 100644 --- a/modules/mesh-delivery-checker/internal/check/examiner_test.go +++ b/modules/mesh-delivery-checker/internal/check/examiner_test.go @@ -15,6 +15,7 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-sdk/go/buildstep" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" ) @@ -108,6 +109,7 @@ const digD = "dd44dd44dd44dd44dd44dd44dd44dd44dd44dd44dd44dd44dd44dd44dd44dd44" type bench struct { ex *Examiner + asks *buildrun.FakeAsks steps *fakeSteps kept *memKV prev *memKV @@ -116,6 +118,50 @@ type bench struct { test func(c stepCall) (any, error) } +// refsOf reads the siblings of a fetch's arguments, as they arrive in an ask (JSON) or as the checker gave them. +func refsOf(v any) []BesideRef { + raw, _ := json.Marshal(v) + var refs []BesideRef + _ = json.Unmarshal(raw, &refs) + return refs +} + +// toolPrefix is the start of each build module's tool names, as the fake build modules answer them. +var toolPrefix = map[string]string{buildrun.GoBuildModule: "build_go_", buildrun.NodeBuildModule: "build_node_"} + +// answer runs a check ask as the build module that pulls it does (buildstep.RunAsk), with the fake tools on shanks and the +// bench's kept verdicts, and answers the step it writes. +func (b *bench) answer(a delivery.StepAsk) delivery.Step { + module, _ := delivery.BuildModuleOf(a.Language) + const node = "shanks" + tool := func(name string) buildstep.StepFunc { + return func(ctx context.Context, args map[string]any) (any, error) { + raw, err := b.steps.Run(ctx, module, node, toolPrefix[module]+name, args) + if err != nil { + return nil, err + } + return raw, nil + } + } + steps := buildstep.AskSteps{Fetch: tool("fetch"), Test: tool("test"), + Kept: func(ctx context.Context, level, digest string) (int, bool, error) { + k, found, err := (Kept{Store: b.kept}).Lookup(ctx, level, digest) + return k.Tests, found, err + }} + if module == buildrun.GoBuildModule { + steps.Digest = tool("digest") + } + out, err := buildstep.RunAsk(context.Background(), steps, a) + began := time.Date(2026, 10, 21, 19, 0, 0, 0, time.UTC) + step := delivery.Step{ID: a.ID, Tool: a.Kind + " ask", Node: node, State: delivery.StepPassed, Started: began, Ended: began.Add(2 * time.Second)} + if err != nil { + step.State, step.Reason = delivery.StepFailed, err.Error() + return step + } + step.Result, _ = json.Marshal(out) + return step +} + // newBench is an examiner over fake build modules that merge cleanly, digest the captured fixture's packages and pass // the captured slow fixture's result at both levels, on shanks. func newBench(t testing.TB) *bench { @@ -142,7 +188,8 @@ func newBench(t testing.TB) *bench { "build_node_test": func(c stepCall) (any, error) { return b.test(c) }, } now := time.Date(2026, 10, 21, 19, 0, 0, 0, time.UTC) - b.ex = &Examiner{Steps: b.steps, Toolchains: goEntries("shanks"), Places: &buildrun.Places{}, Kept: Kept{Store: b.kept}, Previews: b.prev, + b.asks = &buildrun.FakeAsks{Answer: b.answer} + b.ex = &Examiner{Asks: buildrun.Asker{Publisher: b.asks, Waiter: b.asks, Bound: 5 * time.Second}, Toolchains: goEntries("shanks"), Kept: Kept{Store: b.kept}, Previews: b.prev, Modules: langs{}, Trunks: &Trunks{}, Now: func() time.Time { now = now.Add(50 * time.Millisecond); return now }} return b } @@ -353,33 +400,6 @@ func TestAHeadThatDoesNotMergeEndsFailedNamingTheFilesAndRunsNoTest(t *testing.T } } -func TestNoQualifyingNodeIsAFailureOfTheNodes(t *testing.T) { // N6/AC4 - b := newBench(t) - b.ex.Toolchains = entries{} - out := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))) - if !IsNodesFailure(out.Verdict) { - t.Fatalf("verdict = %+v, want a failure of the nodes", out.Verdict) - } - b.ex.Toolchains = goEntries("shanks") - b.ex.Presence = offline{} - if out = b.judge(t, caseOf(t, pullOf(repoA, 7, head1))); !IsNodesFailure(out.Verdict) { - t.Fatalf("an offline node: verdict = %+v", out.Verdict) - } -} - -type offline struct{} - -func (offline) Online(string) bool { return false } - -func TestAStepThatFailsIsAVerdictNamingTheNode(t *testing.T) { - b := newBench(t) - b.steps.handlers["build_go_fetch"] = func(stepCall) (any, error) { return nil, fmt.Errorf("the step failed: ssh refused") } - out := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))) - if f := out.Verdict.Failed; out.Verdict.State != delivery.VerdictFail || f.Machine != "shanks" || !strings.Contains(f.Words, "ssh refused") { - t.Fatalf("verdict = %+v", out.Verdict) - } -} - func TestALevelWhoseDigestIsNotNamedRunsWholeAndKeepsNothing(t *testing.T) { // 1287/AC4 b := newBench(t) b.steps.handlers["build_go_digest"] = func(stepCall) (any, error) { return nil, fmt.Errorf("go list failed") } @@ -441,7 +461,7 @@ func TestAGroupIsOneVerdictAndEachMembersTestsReadTheOthersAtTheirHeads(t *testi var forA []BesideRef for _, f := range fetch { if f.args["repository"] == repoA { - forA, _ = f.args["beside"].([]BesideRef) + forA, _ = refsOf(f.args["beside"]), true } } found := false @@ -599,7 +619,7 @@ func TestNothingIsKeptForAWorkspaceGivenSiblingsWhenTheDigestDoesNotSayItCoversT func echoBeside(b *bench, d Digested) Digested { var lines []string for _, f := range b.steps.tool("fetch") { - refs, _ := f.args["beside"].([]BesideRef) + refs, _ := refsOf(f.args["beside"]), true for _, r := range refs { lines = append(lines, r.Repository+" "+r.Commit) } @@ -614,7 +634,7 @@ func besideDigest(b *bench) func(c stepCall) (any, error) { level := c.args["level"].(string) var lines []string for _, f := range b.steps.tool("fetch") { - refs, _ := f.args["beside"].([]BesideRef) + refs, _ := refsOf(f.args["beside"]), true for _, r := range refs { lines = append(lines, r.Repository+" "+r.Commit) } @@ -741,40 +761,37 @@ func TestNoStepOfACheckCarriesPublish(t *testing.T) { // ADR 0500 §4, 1309 (the } } -// keyCaller is a build module that starts every long step and answers its key on the node it was asked on. -type keyCaller struct{} - -func (keyCaller) CallOn(node, key string, _ any) (json.RawMessage, error) { - return json.Marshal(map[string]string{"step": "lost1", "key": node + ".lost1", "state": delivery.StepRunning}) -} - -// lostSteps is a steps state whose key never ends: the build module stopped, restarted or was unassigned with the step -// running, and nothing writes its end. -type lostSteps struct{} - -func (lostSteps) Wait(ctx context.Context, _, _ string) (delivery.Step, error) { - <-ctx.Done() - return delivery.Step{}, ctx.Err() -} - -func TestAStepThatNeverEndsEndsTheCheckLoudlyAndFreesItsNodesPlace(t *testing.T) { // 1890/AC1 +// 1890/AC1, 1887/A5a: an ask whose step never ends (the build node stopped with it running, and nothing redelivered it in +// time) ends the check loudly at the asker's bound; the checker holds no place to free. +func TestAnAskWhoseStepNeverEndsEndsTheCheckLoudly(t *testing.T) { b := newBench(t) - places := &buildrun.Places{Capacity: map[string]int{"shanks": 1}} - b.ex.Places = places - b.ex.Steps = buildrun.Runner{Caller: keyCaller{}, Steps: lostSteps{}, Bound: 20 * time.Millisecond} + b.asks.Answer = nil + b.ex.Asks = buildrun.Asker{Publisher: b.asks, Waiter: b.asks, Bound: 20 * time.Millisecond} out := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))) - if f := out.Verdict.Failed; out.Verdict.State != delivery.VerdictFail || f.Machine != "shanks" || !strings.Contains(f.Words, "shanks.lost1 did not end within") { + if f := out.Verdict.Failed; out.Verdict.State != delivery.VerdictFail || !strings.Contains(f.Words, "did not end within") { t.Fatalf("verdict = %+v", out.Verdict) } - if r, res := places.Running(), places.Reserved(); len(r) != 0 || len(res) != 0 { - t.Fatalf("the lost step still holds shanks: running %v, reserved %v", r, res) + if n := len(b.asks.Asks()); n != 1 || b.asks.Asks()[0].Kind != delivery.AskCheck || b.asks.Asks()[0].Check == nil { + t.Fatalf("asks = %+v, want one check ask", b.asks.Asks()) + } +} + +// 1887/A5a, ADR 0536 §4: a check is one ask per pull request and build module on the check stream, a re-judge after a +// trunk move one on the recheck stream, each carrying both levels with the kept verdicts taken by the build module. +func TestACheckIsOneAskAndAReJudgeGoesOnTheRecheckStream(t *testing.T) { + b := newBench(t) + b.judge(t, caseOf(t, pullOf(repoA, 7, head1))) + asks := b.asks.Asks() + if len(asks) != 1 || asks[0].Kind != delivery.AskCheck || asks[0].Language != delivery.AskGo || asks[0].Asker != delivery.CheckerModule { + t.Fatalf("asks = %+v, want one Go check ask", asks) + } + if j := asks[0].Check; j == nil || !j.Kept || strings.Join(j.Levels, ",") != "unit,integration" || j.Fetch["repository"] != repoA { + t.Fatalf("the job = %+v", asks[0].Check) + } + cs := caseOf(t, pullOf(repoA, 7, head1)) + cs.Kind = KindTrunk + b.judge(t, cs) + if asks := b.asks.Asks(); len(asks) != 2 || asks[1].Kind != delivery.AskRecheck { + t.Fatalf("asks = %+v, want the re-judge on recheck", asks) } - // The node's one place is free: the next check takes it at once. - ctx, cancel := context.WithTimeout(context.Background(), time.Second) - defer cancel() - release, err := places.Acquire(ctx, "shanks") - if err != nil { - t.Fatalf("shanks's place was not freed: %v", err) - } - release() } diff --git a/modules/mesh-delivery-checker/internal/check/heard.go b/modules/mesh-delivery-checker/internal/check/heard.go index d97c17f..b10bd09 100644 --- a/modules/mesh-delivery-checker/internal/check/heard.go +++ b/modules/mesh-delivery-checker/internal/check/heard.go @@ -1,18 +1,14 @@ 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. +// bound. The program wraps the errors of its forge calls in it where the bus gives a timeout; a no-responders error is +// known as it is. A build step is no call since R4g: an ask whose node stops is delivered again by its stream (ADR 0525 §6). type Unheard struct{ Err error } func (u *Unheard) Error() string { return u.Err.Error() } @@ -23,79 +19,3 @@ 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 index 784ffc7..3072d8c 100644 --- a/modules/mesh-delivery-checker/internal/check/heard_test.go +++ b/modules/mesh-delivery-checker/internal/check/heard_test.go @@ -1,9 +1,7 @@ package check import ( - "context" "fmt" - "strings" "testing" "github.com/nats-io/nats.go" @@ -11,127 +9,8 @@ import ( "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) - } -} - +// Issue 1820 AC5: a read of the forge that nobody answers is the forge's failure, not a build node's. (AC1-AC4 moved a +// step to another node; since R4g an ask whose node stops is delivered again by its stream: ADR 0525 §6.) 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) diff --git a/modules/mesh-delivery-checker/internal/check/judge.go b/modules/mesh-delivery-checker/internal/check/judge.go index dcd07f0..9351d50 100644 --- a/modules/mesh-delivery-checker/internal/check/judge.go +++ b/modules/mesh-delivery-checker/internal/check/judge.go @@ -10,6 +10,7 @@ import ( "strings" "time" + "git.novox.be/novox/mesh-sdk/go/buildstep" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" ) @@ -51,9 +52,11 @@ type verdictPtr = *delivery.Verdict func failed(machine, words string) verdictPtr { return fail(machine, words) } -// Steps runs a build module's long step on a node and answers its result (buildrun.Runner). -type Steps interface { - Run(ctx context.Context, module, node, tool string, args map[string]any) (json.RawMessage, error) +// Asks puts a check's ask on its build seat's stream and waits for its step (buildrun.Asker; ADR 0525 §1, §6): the +// checker chooses no node, starts no step and holds none; whichever build node pulls the ask runs the check. +type Asks interface { + Put(ctx context.Context, a delivery.StepAsk) error + Wait(ctx context.Context, a delivery.StepAsk) (delivery.Step, error) } // ToolchainEntries are the nodes' toolchain entries of a language (buildrun.ToolchainSet). @@ -67,24 +70,18 @@ type BuildModules interface { BuildModules(repository string) []string } -// toolPrefix is the start of each build module's tool names. -var toolPrefix = map[string]string{buildrun.GoBuildModule: "build_go_", buildrun.NodeBuildModule: "build_node_"} +// askLanguage is the build seat language of each build module the checker asks. +var askLanguage = map[string]string{buildrun.GoBuildModule: delivery.AskGo, buildrun.NodeBuildModule: delivery.AskNode} -// Examiner judges a case through the build modules (ADR 0484 clauses 3, 4, 10, 13): for each pull request it fetches the head -// merged onto its trunk head on a build node, then runs the unit level and the integration level, taking the kept verdicts -// of unchanged packages, running the change's own tests first and stopping at the first failure. +// Examiner judges a case through the build seats (ADR 0484 clauses 3, 4, 10, 13; ADR 0525 §1, §3): for each pull request +// and build module it puts one check ask whose job fetches the head merged onto its trunk head, then runs the unit level +// and the integration level, taking the kept verdicts of unchanged packages, running the change's own tests first and +// stopping at the first failure; it judges the step's answer as it judged the tools' answers. type Examiner struct { - Steps Steps + Asks Asks + // Toolchains are the nodes' toolchains: a kept verdict names the toolchain the nodes agree on. Toolchains ToolchainEntries - // Presence says which nodes are online; Health whether a node's build module is running and healthy. A node that is not - // is not chosen; nil means every node is. - Presence interface{ Online(node string) bool } - Health interface { - Healthy(node, module string) bool - } - // Places holds a step back while its node is at the checker's capacity (ADR 0484 clause 12). - Places *buildrun.Places - Kept Kept + Kept Kept // Previews is the previews bucket; nil writes none. Previews KV Modules BuildModules @@ -132,22 +129,22 @@ func besideLines(refs []BesideRef) []string { return out } -// target is a pull request on one build module: the node chosen, the workspace its fetch made and what it said. +// target is a pull request on one build module: the node that pulled its ask, the workspace its fetch made and what it +// said. type target struct { pull Pull module string 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 are the arguments of the fetch its check ask runs. 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 are the lines ` ` of the siblings the workspace was given, sorted: its tests may read them, // so a verdict is kept or taken only under a digest that says it folded exactly these in (lead decision L5). beside []string - plans map[string]*goPlan // by level - // runner is RunnerWorkspace while the target's steps are asked to run the candidate (ADR 0535 §1), empty otherwise. + // runner is RunnerWorkspace when the target's ask runs the candidate's own build module (ADR 0535 §1), empty otherwise. runner string // judged is what the line says of the program that judged a change to the build module; empty for any other change. judged string @@ -177,14 +174,6 @@ func (t *target) stepArgs(args map[string]any) map[string]any { return args } -// goPlan is what a Go target has to run at one level: the digests, the packages without a kept verdict and the tests taken. -type goPlan struct { - digests map[string]string - todo []string - // all says the digest step gave nothing to key a kept verdict by, so the level runs whole and keeps nothing. - all bool -} - // acc collects a level's counts across targets. type acc struct { run, kept, skipped int @@ -200,7 +189,6 @@ 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 } @@ -209,24 +197,9 @@ type state struct { // be judged at all. func (j *Examiner) Judge(ctx context.Context, c *Case) (Outcome, error) { st := &state{j: j, c: c, levels: map[string]*acc{delivery.LevelUnit: {}, delivery.LevelIntegration: {}}, nodes: map[string]bool{}} - if v := st.fetchAll(ctx); v != nil { + if v := st.judgeAll(ctx); v != nil { return st.finish(ctx, *v), nil } - if v := st.checkPins(ctx); v != nil { - return st.finish(ctx, *v), nil - } - for _, level := range []string{delivery.LevelUnit, delivery.LevelIntegration} { - if c.Try != nil && c.Try.Level != "" && c.Try.Level != level { - continue // a try of one level runs that level alone - } - for _, priority := range []bool{true, false} { - for _, t := range st.targets { - if v := st.runTarget(ctx, t, level, priority); v != nil { - return st.finish(ctx, *v), nil - } - } - } - } return st.finish(ctx, delivery.Verdict{State: delivery.VerdictPass}), nil } @@ -291,55 +264,18 @@ 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. 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) - } +// levels are the levels a case's asks run: both, or the one a try names. +func (st *state) levelsOf() []string { + if st.c.Try != nil && st.c.Try.Level != "" { + return []string{st.c.Try.Level} // a try of one level runs that level alone } + return []string{delivery.LevelUnit, delivery.LevelIntegration} } -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 { - return nil, err - } - defer release() - return st.j.Steps.Run(ctx, t.module, t.node, toolPrefix[t.module]+tool, args) -} - -// choose picks the build node for a pull request's repository on a build module. -func (st *state) choose(bm string) (node, toolchain string, err error) { - language := buildrun.LanguageOf[bm] - qualifies := func(node string) bool { - 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 { - return "", "", err - } - return nodes[0], tc.Version, nil -} - -// fetchAll fetches every pull request's head merged onto its trunk on a build node of each build module the repository -// needs. A head that does not merge, or a step that fails, ends the check at once, running no test (ADR 0484 clause 13). -func (st *state) fetchAll(ctx context.Context) *delivery.Verdict { - began := st.j.now() - defer func() { st.clone = st.j.now().Sub(began) }() +// judgeAll puts one check ask per pull request and build module the repository needs, the mesh-sdk member first, and +// judges each answer. A head that does not merge, a pin that is not allowed or a level that fails ends the check at once +// (ADR 0484 clause 13). +func (st *state) judgeAll(ctx context.Context) *delivery.Verdict { // The mesh-sdk member first: whether its merge changes the TypeScript package's paths decides whether the other members' // Node fetches take the group's package (ADR 0500 §5). pulls := slices.Clone(st.c.Pulls) @@ -356,47 +292,32 @@ func (st *state) fetchAll(ctx context.Context) *delivery.Verdict { for _, p := range pulls { var merged []Fetched for _, bm := range st.j.Modules.BuildModules(p.Repository) { - node, tc, err := st.choose(bm) - if err != nil { - return fail(bm, err.Error()) + t := &target{pull: p, module: bm} + if tc, none := delivery.AgreedToolchain(st.j.Toolchains.Entries(buildrun.LanguageOf[bm])); none == "" { + t.toolchain = tc.Version } - 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.NodeBuildModule && groupPackage != nil && p.Repository != SDKRepository { args["package"] = groupPackage } if bm == buildrun.GoBuildModule { beside, err := st.j.besideFor(ctx, st.c, p) if err != nil { - return fail(forgeMachine(err, node), err.Error()+unheardWords(err)) + return fail(forgeMachine(err, bm), err.Error()+unheardWords(err)) } if len(beside) > 0 { args["beside"] = beside t.beside = besideLines(beside) } - } - 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()))) - } - if err := json.Unmarshal(raw, &t.fetched); err != nil || t.fetched.Workspace == "" { - return fail(node, fmt.Sprintf("the fetch of %s#%d answered no workspace: %s", p.Repository, p.Number, raw)) - } - if len(t.fetched.Conflicts) > 0 { - if st.c.Try == nil { - st.j.writePreview(ctx, p, append(merged, t.fetched)) + if p.Repository == catalogueRepository { + t.runner = RunnerWorkspace // used by the build module when the merge changes it (ADR 0535 §1) } - return fail(node, fmt.Sprintf("%s#%d does not merge into the trunk: files in conflict: %s", p.Repository, p.Number, strings.Join(t.fetched.Conflicts, ", "))) } - t.changed = t.fetched.Changed - if bm == buildrun.GoBuildModule && changesBuildGo(p.Repository, t.changed) { - t.runner = RunnerWorkspace - } - merged = append(merged, t.fetched) + t.fetchArgs = args st.targets = append(st.targets, t) + if v := st.ask(ctx, t, &merged); v != nil { + return v + } } if st.c.Try == nil { // a try owes the indexer no preview: no pull request holds its head st.j.writePreview(ctx, p, merged) @@ -408,6 +329,138 @@ func (st *state) fetchAll(ctx context.Context) *delivery.Verdict { return nil } +// ask puts the target's check ask, waits for the step of the node that pulled it and judges its answer: the fetch, the +// pins of its go.mod files, then each level's digest, kept verdicts and tests. +func (st *state) ask(ctx context.Context, t *target, merged *[]Fetched) *delivery.Verdict { + p := t.pull + job := &delivery.CheckJob{Fetch: t.fetchArgs, Levels: st.levelsOf(), Kept: true} + if t.runner != "" { + job.Runner, job.RunnerWhen = t.runner, buildGoDir + } + if st.c.Try != nil && st.c.Try.Run != "" { + job.Run, job.Kept = st.c.Try.Run, false // a try of some tests runs them alone: a partial run proves no package + } + kind := delivery.AskCheck + if st.c.Kind == KindTrunk { + kind = delivery.AskRecheck // a re-judge after a trunk move goes after every new head's check (ADR 0536 §4) + } + a := delivery.StepAsk{ID: buildstep.NewID(), Kind: kind, Language: askLanguage[t.module], Asker: delivery.CheckerModule, + At: st.j.now().UTC(), Check: job} + st.c.Progress.Set(fmt.Sprintf("the %s ask %s of %s#%d", kind, a.ID, p.Repository, p.Number), "") + if err := st.j.Asks.Put(ctx, a); err != nil { + return fail(t.module, fmt.Sprintf("the check of %s#%d could not be asked: %v", p.Repository, p.Number, err)) + } + step, err := st.j.Asks.Wait(ctx, a) + if err != nil { + if ctx.Err() != nil { + return fail(t.module, ctx.Err().Error()) + } + return fail(t.module, fmt.Sprintf("the check of %s#%d: %v", p.Repository, p.Number, err)) + } + // The fetch and the digests are what of the ask's time its levels' running did not take (Took.Clone). + took := time.Duration(0) + if !step.Started.IsZero() && step.Ended.After(step.Started) { + took = step.Ended.Sub(step.Started) + } + t.node = step.Node + st.nodes[step.Node] = true + st.c.Progress.Set(fmt.Sprintf("judge the check of %s#%d", p.Repository, p.Number), step.Node) + if step.State != delivery.StepPassed { + return fail(step.Node, fmt.Sprintf("the check of %s#%d %s: %s%s", p.Repository, p.Number, step.State, step.Reason, pinHint(st.c, step.Reason))) + } + var answer delivery.CheckAnswer + if err := json.Unmarshal(step.Result, &answer); err != nil { + return fail(step.Node, fmt.Sprintf("the check of %s#%d answered what is not a check: %s", p.Repository, p.Number, step.Result)) + } + if err := json.Unmarshal(answer.Fetch, &t.fetched); err != nil || t.fetched.Workspace == "" { + return fail(step.Node, fmt.Sprintf("the fetch of %s#%d answered no workspace: %s", p.Repository, p.Number, answer.Fetch)) + } + if len(t.fetched.Conflicts) > 0 { + if st.c.Try == nil { + st.j.writePreview(ctx, p, append(*merged, t.fetched)) + } + return fail(step.Node, fmt.Sprintf("%s#%d does not merge into the trunk: files in conflict: %s", p.Repository, p.Number, strings.Join(t.fetched.Conflicts, ", "))) + } + t.changed = t.fetched.Changed + if !changesBuildGo(p.Repository, t.changed) { + t.runner = "" // the build module used no runner: the merge does not change it + } + *merged = append(*merged, t.fetched) + if v := st.checkPinsOf(ctx, t); v != nil { + return v + } + ran := 0.0 + for _, la := range answer.Levels { + v, elapsed := st.level(ctx, t, la) + ran += elapsed + if v != nil { + st.clone += max(0, took-time.Duration(ran*float64(time.Second))) + return v + } + } + st.clone += max(0, took-time.Duration(ran*float64(time.Second))) + return nil +} + +// level judges one level of a check ask's answer as the tools' answers were judged: the digest keys the kept verdicts +// of the packages that passed, the kept ones taken are counted, and the first test result that did not pass is the +// verdict. It answers the level's running time (its tests' and setup's). +func (st *state) level(ctx context.Context, t *target, la delivery.LevelAnswer) (*delivery.Verdict, float64) { + level := la.Level + a := st.levels[level] + if a == nil { + return fail(t.node, fmt.Sprintf("the check of %s#%d answered the level %q, which is no level", t.pull.Repository, t.pull.Number, level)), 0 + } + var digests map[string]string + if t.module == buildrun.GoBuildModule && len(la.Digest) > 0 { + var d Digested + if err := json.Unmarshal(la.Digest, &d); err == nil { + digests = d.Packages + if t.runner != "" { + t.judgedBy(d.JudgedBy) // the installed module judged it: the runner is not used again + } + } + } + if t.runner != "" && la.Judged != "" { + t.judgedBy(la.Judged) // the candidate judged the level: nothing is kept (ADR 0535 §4) + } + for _, k := range la.Kept { + a.kept += k.Tests + } + elapsed := 0.0 + for _, raw := range la.Tests { + var res Tested + if err := json.Unmarshal(raw, &res); err != nil { + return fail(t.node, fmt.Sprintf("the %s level of %s#%d answered what is not a result: %v", level, t.pull.Repository, t.pull.Number, err)), elapsed + } + if t.runner != "" { + t.judgedBy(res.JudgedBy) + } + elapsed += res.Elapsed + res.Setup + if t.module == buildrun.NodeBuildModule { + if !res.NoTests { + a.run++ // a Node package is counted whole: its build module counts no tests (ADR 0484 clause 10) + } + } else { + a.run += res.Tests + a.skipped += res.Skipped + a.slowest = append(a.slowest, res.Slowest...) + } + a.elapsed += res.Elapsed + a.setup += res.Setup + a.over = a.over || res.OverBound + if len(digests) > 0 { + if _, err := st.j.Kept.KeepPasses(ctx, t.pull.Repository, level, t.toolchain, res, digests, st.j.now()); err != nil { + st.j.logf("%s#%d: a passing package could not be kept: %v", t.pull.Repository, t.pull.Number, err) + } + } + if !res.Passed { + return levelFailure(t, level, res), elapsed + } + } + return nil, elapsed +} + // changesPackage says whether the merge of a pull request changes a file inside the named paths of one of its repository's // TypeScript packages: a change of the manifest alone, or of files outside those paths, does not (ADR 0500 §5). func (j *Examiner) changesPackage(p Pull, merged []Fetched) bool { @@ -428,51 +481,6 @@ func (j *Examiner) changesPackage(p Pull, merged []Fetched) bool { return false } -// isTestFile says whether a repository-relative path is a test file of the languages the checker judges: Go's _test.go and -// the TypeScript module's .test. and .spec. files and test directories. -func isTestFile(p string) bool { - switch { - case strings.HasSuffix(p, "_test.go"): - return true - case strings.Contains(p, ".test.") || strings.Contains(p, ".spec."): - return true - } - for _, seg := range strings.Split(path.Dir(p), "/") { - if seg == "__tests__" || seg == "test" || seg == "tests" { - return true - } - } - return false -} - -// changesTestOf says whether the merge changes a test file of the Go package with the import path pkg: the directory of the -// file and the import path agree on their last segments, whichever holds the module's directory and path in front. -func changesTestOf(changed []string, pkg string) bool { - for _, f := range changed { - if !strings.HasSuffix(f, "_test.go") { - continue - } - dir := path.Dir(f) - if dir == pkg || strings.HasSuffix(pkg, "/"+dir) || strings.HasSuffix(dir, "/"+pkg) { - return true - } - } - return false -} - -// changesTestIn says whether the merge changes a test file inside the directory of a Node module. -func changesTestIn(changed []string, dir string) bool { - for _, f := range changed { - if !isTestFile(f) || strings.HasSuffix(f, "_test.go") { - continue - } - if dir == "." || dir == "" || strings.HasPrefix(f, strings.TrimSuffix(dir, "/")+"/") { - return true - } - } - return false -} - // failedName names a failure: its package and test. func failedName(f Failure) string { if f.Test == "" { @@ -514,161 +522,3 @@ func levelFailure(t *target, level string, res Tested) *delivery.Verdict { } return &v } - -// runTarget runs one level of a target: first the packages the change's own tests are in (priority), then the others. -func (st *state) runTarget(ctx context.Context, t *target, level string, priority bool) *delivery.Verdict { - if t.module == buildrun.NodeBuildModule { - return st.runNode(ctx, t, level, priority) - } - return st.runGo(ctx, t, level, priority) -} - -// A level's work is planned once for a target: the digest of each package, the kept verdicts taken, the packages left to run. -func (st *state) planGo(ctx context.Context, t *target, level string) *delivery.Verdict { - if t.plans[level] != nil { - return nil - } - plan := &goPlan{digests: map[string]string{}} - t.plans[level] = plan - if st.c.Try != nil && st.c.Try.Run != "" { - // A try of some tests runs them alone: a partial run proves no package, so nothing is taken or kept. - plan.all = true - return nil - } - raw, err := st.step(ctx, t, "digest", fmt.Sprintf("digest the %s level of %s#%d on %s", level, t.pull.Repository, t.pull.Number, t.node), - t.stepArgs(map[string]any{"workspace": t.fetched.Workspace, "level": level})) - var d Digested - if err == nil { - err = json.Unmarshal(raw, &d) - } - if ctx.Err() != nil { - return fail(t.node, ctx.Err().Error()) - } - if err == nil && t.runner != "" && t.judgedBy(d.JudgedBy) { - // The candidate's digest decides which verdicts would be taken: every package runs and nothing is kept (ADR 0535 §4). - plan.all = true - return nil - } - if err != nil || len(d.Packages) == 0 { - // The build module named no digest (a level it cannot digest here): the level runs whole and keeps nothing. - if err != nil { - st.j.logf("%s#%d: no digest of the %s level, so it runs whole: %v", t.pull.Repository, t.pull.Number, level, err) - } - plan.all = true - return nil - } - if !slices.Equal(d.Beside, t.beside) { - // The siblings the tests read are not all in the digest (a build module before DigestVersion 3, or another set): a - // package whose tests read one is never kept, nor taken from a verdict kept for other inputs (ADR 0444, L5). - st.j.logf("%s#%d: the %s level's digest covers the siblings %v, not %v, so it runs whole and keeps nothing", t.pull.Repository, t.pull.Number, level, d.Beside, t.beside) - plan.all = true - return nil - } - plan.digests = d.Packages - a := st.levels[level] - pkgs := make([]string, 0, len(d.Packages)) - for p := range d.Packages { - pkgs = append(pkgs, p) - } - sort.Strings(pkgs) - for _, p := range pkgs { - k, found, err := st.j.Kept.Lookup(ctx, level, d.Packages[p]) - if err != nil { - st.j.logf("the kept verdict of %s at %s cannot be read, so it runs: %v", p, level, err) - } - if found { - a.kept += k.Tests - continue - } - plan.todo = append(plan.todo, p) - } - return nil -} - -func (st *state) runGo(ctx context.Context, t *target, level string, priority bool) *delivery.Verdict { - if v := st.planGo(ctx, t, level); v != nil { - return v - } - plan := t.plans[level] - var run []string - switch { - case plan.all: - if priority { - return nil // nothing to put first: the level runs whole after - } - default: - for _, p := range plan.todo { - if changesTestOf(t.changed, p) == priority { - run = append(run, p) - } - } - if len(run) == 0 { - return nil - } - } - args := t.stepArgs(map[string]any{"workspace": t.fetched.Workspace, "level": level}) - if run != nil { - args["packages"] = run - } - if st.c.Try != nil && st.c.Try.Run != "" { - args["run"] = st.c.Try.Run // build_go_test's go test -run pattern - } - raw, err := st.step(ctx, t, "test", fmt.Sprintf("test the %s level of %s#%d on %s", level, t.pull.Repository, t.pull.Number, t.node), args) - if err != nil { - return fail(t.node, fmt.Sprintf("the %s level of %s#%d: %v", level, t.pull.Repository, t.pull.Number, err)) - } - var res Tested - if err := json.Unmarshal(raw, &res); err != nil { - return fail(t.node, fmt.Sprintf("the %s level of %s#%d answered what is not a result: %v", level, t.pull.Repository, t.pull.Number, err)) - } - if t.runner != "" { - t.judgedBy(res.JudgedBy) - } - a := st.levels[level] - a.run += res.Tests - a.skipped += res.Skipped - a.elapsed += res.Elapsed - a.setup += res.Setup - a.over = a.over || res.OverBound - a.slowest = append(a.slowest, res.Slowest...) - if len(plan.digests) > 0 { - if _, err := st.j.Kept.KeepPasses(ctx, t.pull.Repository, level, t.toolchain, res, plan.digests, st.j.now()); err != nil { - st.j.logf("%s#%d: a passing package could not be kept: %v", t.pull.Repository, t.pull.Number, err) - } - } - if !res.Passed { - return levelFailure(t, level, res) - } - return nil -} - -func (st *state) runNode(ctx context.Context, t *target, level string, priority bool) *delivery.Verdict { - for _, dir := range t.fetched.Modules { - if changesTestIn(t.changed, dir) != priority { - continue - } - raw, err := st.step(ctx, t, "test", fmt.Sprintf("test the %s level of %s#%d (%s) on %s", level, t.pull.Repository, t.pull.Number, dir, t.node), - map[string]any{"workspace": t.fetched.Workspace, "module": dir, "level": level}) - if err != nil { - return fail(t.node, fmt.Sprintf("the %s level of %s#%d (%s): %v", level, t.pull.Repository, t.pull.Number, dir, err)) - } - var res Tested - if err := json.Unmarshal(raw, &res); err != nil { - return fail(t.node, fmt.Sprintf("the %s level of %s#%d (%s) answered what is not a result: %v", level, t.pull.Repository, t.pull.Number, dir, err)) - } - if res.Module == "" { - res.Module = dir - } - a := st.levels[level] - if !res.NoTests { - a.run++ // a Node package is counted whole: its build module counts no tests (ADR 0484 clause 10) - } - a.elapsed += res.Elapsed - a.setup += res.Setup - a.over = a.over || res.OverBound - if !res.Passed { - return levelFailure(t, level, res) - } - } - return nil -} diff --git a/modules/mesh-delivery-checker/internal/check/pins.go b/modules/mesh-delivery-checker/internal/check/pins.go index f1e2937..c07a8c4 100644 --- a/modules/mesh-delivery-checker/internal/check/pins.go +++ b/modules/mesh-delivery-checker/internal/check/pins.go @@ -55,19 +55,19 @@ func SDKPin(goMod string) (commit string, ok bool) { return "", false } -// checkPins refuses a Go member whose go.mod names a mesh-sdk commit that is neither on mesh-sdk's trunk nor the head of the +// checkPinsOf refuses a Go member whose go.mod names a mesh-sdk commit that is neither on mesh-sdk's trunk nor the head of the // group's mesh-sdk pull request. It reads each go.mod the fetch named from the forge: the one the check builds, which is the // head's when the head holds the file and the trunk's when it does not, since a module added on the trunk after the branch was // cut is in the merge and not at the head (issue 1845). It answers nil when every pin is the group's. -func (st *state) checkPins(ctx context.Context) verdictPtr { +func (st *state) checkPinsOf(ctx context.Context, t *target) verdictPtr { if st.j.Forge == nil { return nil } sdk := st.c.member(SDKRepository) - for _, t := range st.targets { - if t.module != goModule || t.pull.Repository == SDKRepository { - continue - } + if t.module != goModule || t.pull.Repository == SDKRepository { + return nil + } + { for _, dir := range t.fetched.Modules { file := path.Join(dir, "go.mod") st.c.Progress.Set(fmt.Sprintf("read %s of %s#%d", file, t.pull.Repository, t.pull.Number), t.node) diff --git a/modules/mesh-delivery-checker/internal/check/pins_test.go b/modules/mesh-delivery-checker/internal/check/pins_test.go index 9d0c6bb..ee8f873 100644 --- a/modules/mesh-delivery-checker/internal/check/pins_test.go +++ b/modules/mesh-delivery-checker/internal/check/pins_test.go @@ -97,7 +97,7 @@ func TestAMemberPinnedToTheSDKMembersHeadPassesAndIsBuiltAgainstIt(t *testing.T) var besideSDK bool for _, f := range b.steps.tool("fetch") { if f.args["repository"] == repoA { - refs, _ := f.args["beside"].([]BesideRef) + refs, _ := refsOf(f.args["beside"]), true for _, r := range refs { besideSDK = besideSDK || (r.Repository == repoSDK && r.Commit == head2) } @@ -116,9 +116,8 @@ func TestAMemberPinnedToAnOlderHeadOfTheSDKMemberFailsNamingTheMemberAndTheCommi !strings.Contains(f.Words, "neither on mesh-sdk's trunk nor the group's head") || !strings.Contains(f.Words, short12(head2)) { t.Fatalf("verdict = %+v", v) } - if n := len(b.steps.tool("test")); n != 0 { - t.Errorf("%d test steps ran after a stale pin", n) - } + // Since R4g the levels run in the member's one check ask before its go.mod pins are judged (ADR 0525 §3); the verdict is + // still the pin's, whatever the levels said. } func TestAMemberPinnedToACommitOnTheTrunkPasses(t *testing.T) { // 1624/AC2, AC4 @@ -213,7 +212,7 @@ func besideOf(t *testing.T, b *bench, repository string) map[string]BesideRef { out := map[string]BesideRef{} for _, f := range b.steps.tool("fetch") { if f.args["repository"] == repository { - refs, _ := f.args["beside"].([]BesideRef) + refs, _ := refsOf(f.args["beside"]), true for _, r := range refs { out[r.Repository] = r } @@ -228,7 +227,8 @@ func TestTheSiblingsAreWhereTheTestsStateThem(t *testing.T) { // review D2, ADR b := newBench(t) b.ex.Trunks = trunksOf(t, map[string]string{controller: trunkA, catalogue: trunkS, host: trunkA2, SDKRepository: "9999999999999999999999999999999999999999"}) b.ex.Runs = runsAt{"mesh-host": hostRuns} - forge := &fakeForge{goMods: map[string]string{controller + " go.mod " + head1: capturedGoMod(t, "cb54756c3865")}, onTrunk: map[string]bool{"cb54756c3865": true}} + forge := &fakeForge{goMods: map[string]string{controller + " go.mod " + head1: capturedGoMod(t, "cb54756c3865"), + catalogue + " go.mod " + head2: "module git.novox.be/novox/mesh-catalog\n"}, onTrunk: map[string]bool{"cb54756c3865": true}} b.ex.Forge = forge // A group of the controller and the catalogue: the controller's tests read the catalogue's trunk, not the group's head. b.judge(t, caseOf(t, pullOf(controller, 7, head1), pullOf(catalogue, 12, head2))) diff --git a/modules/mesh-delivery-checker/internal/check/try_test.go b/modules/mesh-delivery-checker/internal/check/try_test.go index 65b89b5..5c9e5a0 100644 --- a/modules/mesh-delivery-checker/internal/check/try_test.go +++ b/modules/mesh-delivery-checker/internal/check/try_test.go @@ -153,18 +153,7 @@ func TestAFailingTryNamesWhatFailedAndAStepThatCannotRunIsAnError(t *testing.T) if got := readTry(t, kv, id); got.State != TryFail || len(got.Failed) != 1 || !strings.Contains(got.Failed[0], "TestThreeIsFour") { t.Errorf("try = %+v", got) } - // No build node: the try could not run, which is an error, not a failure of the change. - kv2 := newMemKV() - b2 := newBench(t) - b2.ex.Toolchains = entries{} - id2, _ := AskTry(memTries{kv2}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, time.Now()) - tr2 := &Tryer{Tries: kv2, Judge: b2.ex} - raw, _, _, _ = kv2.Get(ctx, id2) - tr2.Put(ctx, id2, raw) - tr2.Wait() - if got := readTry(t, kv2, id2); got.State != TryError { - t.Errorf("try with no node = %+v, want error", got) - } + // No build node is no error since R4g: the try's ask waits in the check stream until a node pulls it (ADR 0525 §4). } func TestATryWaitsWhileADeliveryBuildsAndATryLeftRunningIsTakenOver(t *testing.T) {