diff --git a/modules/mesh-delivery-checker/internal/check/ask_all_test.go b/modules/mesh-delivery-checker/internal/check/ask_all_test.go new file mode 100644 index 0000000..b96e2c6 --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/ask_all_test.go @@ -0,0 +1,61 @@ +package check + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// countingAsks puts asks through inner, and answers no wait until want asks are put (or a bound passes, when it says +// how many were): a group whose asks are put one after another, each waited for, never gets past its first. +type countingAsks struct { + inner Asks + want int + mu sync.Mutex + puts int + all chan struct{} +} + +func (c *countingAsks) Put(ctx context.Context, a delivery.StepAsk) error { + if err := c.inner.Put(ctx, a); err != nil { + return err + } + c.mu.Lock() + c.puts++ + if c.puts == c.want { + close(c.all) + } + c.mu.Unlock() + return nil +} + +func (c *countingAsks) Wait(ctx context.Context, a delivery.StepAsk) (delivery.Step, error) { + select { + case <-c.all: + case <-time.After(time.Second): + c.mu.Lock() + defer c.mu.Unlock() + return delivery.Step{}, fmt.Errorf("%d of %d asks were put before the first was waited for", c.puts, c.want) + } + return c.inner.Wait(ctx, a) +} + +// 1921/S8: a group's asks are all put before any is waited for, so its members run side by side on the build nodes; +// the verdict is judged in member order as before. +func TestAGroupPutsEveryMembersAskBeforeTheFirstAnswer(t *testing.T) { + b := newBench(t) + counting := &countingAsks{inner: b.ex.Asks, want: 3, all: make(chan struct{})} + b.ex.Asks = counting + cs := caseOf(t, pullOf(repoA, 7, head1), pullOf(repoSDK, 12, head2), pullOf("novox/mesh-controller", 3, head1)) + v := b.judge(t, cs).Verdict + if v.State != delivery.VerdictPass { + t.Fatalf("verdict = %+v", v) + } + if counting.puts != 3 { + t.Errorf("%d asks were put, want 3", counting.puts) + } +} diff --git a/modules/mesh-delivery-checker/internal/check/judge.go b/modules/mesh-delivery-checker/internal/check/judge.go index cbf731d..36033b6 100644 --- a/modules/mesh-delivery-checker/internal/check/judge.go +++ b/modules/mesh-delivery-checker/internal/check/judge.go @@ -159,6 +159,9 @@ type target struct { 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 + // ask is the check ask put for it, and failed why it could not be asked (issue 1921, try speed S8). + ask delivery.StepAsk + failed *delivery.Verdict } // judgedBy takes what a step under runner workspace answered of the program that ran it (judged_by): the candidate, or the @@ -307,12 +310,13 @@ func (st *state) levelsOf() []string { return []string{delivery.LevelUnit, delivery.LevelIntegration} } -// 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). +// judgeAll puts one check ask per pull request and build module the repository needs, every ask of the group at once so +// the members run side by side on the build nodes (issue 1921, try speed S8), and judges the answers in member order, the +// mesh-sdk member first. 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): the verdict is the first failure in member order, as when the asks were put one after another. 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). + // Node fetches take the group's package (ADR 0500 §5). Those Node asks alone wait for the mesh-sdk member's answer. pulls := slices.Clone(st.c.Pulls) slices.SortStableFunc(pulls, func(a, b Pull) int { if (a.Repository == SDKRepository) != (b.Repository == SDKRepository) { @@ -323,34 +327,32 @@ func (st *state) judgeAll(ctx context.Context) *delivery.Verdict { } return 0 }) - var groupPackage map[string]any - for _, p := range pulls { - var merged []Fetched + sdkGroup := len(st.c.Pulls) > 1 && len(pulls) > 0 && pulls[0].Repository == SDKRepository + members := make([][]*target, len(pulls)) + var waiting []*target // Node asks of the other members, put once the mesh-sdk member is judged + for i, p := range pulls { for _, bm := range st.j.Modules.BuildModules(p.Repository) { - t := &target{pull: p, module: bm} - if tc, none := delivery.AgreedToolchain(st.j.Toolchains.Entries(buildrun.LanguageOf[bm])); none == "" { - t.toolchain = tc.Version - } - args := map[string]any{"repository": p.Repository, "commit": p.Head, "ref": refOf(p.Number), "base": p.Trunk} - 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, bm), err.Error()+unheardWords(err)) - } - if len(beside) > 0 { - args["beside"] = beside - t.beside = besideLines(beside) - } - if p.Repository == catalogueRepository { - t.runner = RunnerWorkspace // used by the build module when the merge changes it (ADR 0535 §1) - } - } - t.fetchArgs = args + t := st.target(ctx, p, bm) + members[i] = append(members[i], t) st.targets = append(st.targets, t) - if v := st.ask(ctx, t, &merged); v != nil { + if t.failed != nil { + continue // said when the judging reaches it, so the verdict stays the first failure in member order + } + if sdkGroup && bm == buildrun.NodeBuildModule && p.Repository != SDKRepository { + waiting = append(waiting, t) + continue + } + t.failed = st.put(ctx, t) + } + } + var groupPackage map[string]any + for i, p := range pulls { + var merged []Fetched + for _, t := range members[i] { + if t.failed != nil { + return t.failed + } + if v := st.judgeAsk(ctx, t, &merged); v != nil { return v } } @@ -360,13 +362,46 @@ func (st *state) judgeAll(ctx context.Context) *delivery.Verdict { if p.Repository == SDKRepository && len(st.c.Pulls) > 1 && st.j.changesPackage(p, merged) { groupPackage = map[string]any{"repository": p.Repository, "commit": p.Head, "ref": refOf(p.Number)} } + if i == 0 && sdkGroup { + for _, t := range waiting { + if groupPackage != nil { + t.fetchArgs["package"] = groupPackage + } + t.failed = st.put(ctx, t) + } + } } 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 { +// target is the check of one pull request by one build module: its fetch's arguments, its toolchain, the siblings its +// tests read and the runner it is asked with; failed says why it cannot be asked. +func (st *state) target(ctx context.Context, p Pull, bm string) *target { + t := &target{pull: p, module: bm} + if tc, none := delivery.AgreedToolchain(st.j.Toolchains.Entries(buildrun.LanguageOf[bm])); none == "" { + t.toolchain = tc.Version + } + args := map[string]any{"repository": p.Repository, "commit": p.Head, "ref": refOf(p.Number), "base": p.Trunk} + if bm == buildrun.GoBuildModule { + beside, err := st.j.besideFor(ctx, st.c, p) + if err != nil { + t.failed = fail(forgeMachine(err, bm), err.Error()+unheardWords(err)) + return t + } + if len(beside) > 0 { + args["beside"] = beside + t.beside = besideLines(beside) + } + if p.Repository == catalogueRepository { + t.runner = RunnerWorkspace // used by the build module when the merge changes it (ADR 0535 §1) + } + } + t.fetchArgs = args + return t +} + +// put puts the target's check ask on its build seat's stream. +func (st *state) put(ctx context.Context, t *target) *delivery.Verdict { p := t.pull job := &delivery.CheckJob{Fetch: t.fetchArgs, Levels: st.levelsOf(), Kept: true} if t.runner != "" { @@ -379,13 +414,20 @@ func (st *state) ask(ctx context.Context, t *target, merged *[]Fetched) *deliver 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, + t.ask = 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 { + st.c.Progress.Set(fmt.Sprintf("the %s ask %s of %s#%d", kind, t.ask.ID, p.Repository, p.Number), "") + if err := st.j.Asks.Put(ctx, t.ask); 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) + return nil +} + +// judgeAsk waits for the step of the node that pulled the target's ask 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) judgeAsk(ctx context.Context, t *target, merged *[]Fetched) *delivery.Verdict { + p := t.pull + step, err := st.j.Asks.Wait(ctx, t.ask) if err != nil { if ctx.Err() != nil { return fail(t.module, ctx.Err().Error())