diff --git a/modules/mesh-delivery-checker/internal/check/examiner_test.go b/modules/mesh-delivery-checker/internal/check/examiner_test.go index 7f6410d1..2eaee760 100644 --- a/modules/mesh-delivery-checker/internal/check/examiner_test.go +++ b/modules/mesh-delivery-checker/internal/check/examiner_test.go @@ -325,6 +325,59 @@ func packagesOf(c stepCall) []string { return out } +// overlapAsks counts the asks put and not yet ended, and the most at once: a wait holds until a second ask is put, or a +// second passed, so asks put one after another never overlap. +type overlapAsks struct { + Asks + mu sync.Mutex + open, top int + more chan struct{} +} + +func (o *overlapAsks) Put(ctx context.Context, a buildstep.StepAsk) error { + o.mu.Lock() + o.open++ + if o.open > o.top { + o.top = o.open + } + if o.open >= 2 { + select { + case <-o.more: + default: + close(o.more) + } + } + o.mu.Unlock() + return o.Asks.Put(ctx, a) +} + +func (o *overlapAsks) Wait(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, error) { + select { + case <-o.more: + case <-time.After(time.Second): + } + defer func() { o.mu.Lock(); o.open--; o.mu.Unlock() }() + return o.Asks.Wait(ctx, a) +} + +// Design 47 G A-304 (ADR 0525 §4): a phase puts every ask of its own at once and then waits for them all, and the +// verdict is judged in the order of the members as before. +func TestAPhasesAsksAreAllOutstandingAtOnce(t *testing.T) { + b := newBench(t) + o := &overlapAsks{Asks: b.ex.Asks, more: make(chan struct{})} + b.ex.Asks = o + v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1), pullOf(repoSDK, 12, head2))).Verdict + if o.top < 2 { + t.Errorf("at most %d asks were outstanding at once, want both members' asks of the phase", o.top) + } + if v.State != meshdeliverycheckerclient.VerdictPass { + t.Errorf("verdict = %+v", v) + } + if fetch := b.steps.tool("fetch"); len(fetch) != 2 { + t.Errorf("fetch calls = %+v, want one per member", fetch) + } +} + func TestACleanMergeIsFetchedAtItsTrunkAndBothLevelsPass(t *testing.T) { // 1679/AC1 b := newBench(t) cs := caseOf(t, pullOf(repoA, 7, head1)) diff --git a/modules/mesh-delivery-checker/internal/check/judge.go b/modules/mesh-delivery-checker/internal/check/judge.go index 7cdcb208..7a4f9f2b 100644 --- a/modules/mesh-delivery-checker/internal/check/judge.go +++ b/modules/mesh-delivery-checker/internal/check/judge.go @@ -557,30 +557,47 @@ func (st *state) judgeAll(ctx context.Context) { modules := make([][]string, len(pulls)) for i, p := range pulls { modules[i] = st.j.Modules.BuildModules(p.Repository) - for _, bm := range modules[i] { - if st.j.languageOf(bm) == languageTypeScript { - continue + } + // Each phase puts every ask of its own at once and then waits for them all (design 47 G A-304, ADR 0525 §4): its + // asks are judged in the order they were put, whatever order they end in. + phase := func(typeScript bool, group []buildstep.FetchPackage) bool { + var asked []*asking + for i, p := range pulls { + for _, bm := range modules[i] { + if (st.j.languageOf(bm) == languageTypeScript) != typeScript { + continue + } + if stopped() { + return false + } + if t := st.memberTarget(ctx, p, bm, group); t != nil { + if q := st.prepareAsk(ctx, t, &merged[i], st.levelsOf(), nil, nil); q != nil && st.putAsk(ctx, q) { + asked = append(asked, q) + } + } } - if stopped() { - return - } - st.askMember(ctx, p, bm, nil, &merged[i]) } + for _, q := range asked { + go q.wait(ctx, st) + } + for n, q := range asked { + if n > 0 && stopped() { + return false + } + <-q.done + st.judgeAsk(ctx, q) + } + return true + } + if !phase(false, nil) { + return } var group []buildstep.FetchPackage if len(st.c.Pulls) > 1 { group = st.groupPackages(pulls, merged) } - for i, p := range pulls { - for _, bm := range modules[i] { - if st.j.languageOf(bm) != languageTypeScript { - continue - } - if stopped() { - return - } - st.askMember(ctx, p, bm, group, &merged[i]) - } + if !phase(true, group) { + return } for i, p := range pulls { // Every build module's fetch merged (each is in merged once): the preview is written, also when the member's tests @@ -599,9 +616,9 @@ func (st *state) judgeAll(ctx context.Context) { } } -// askMember puts the check ask of one build module of a member and judges its answer; a TypeScript fetch is given the -// group's packages. -func (st *state) askMember(ctx context.Context, p Pull, bm string, group []buildstep.FetchPackage, merged *[]Fetched) { +// memberTarget is the target of the check ask of one build module of a member, nil when it cannot be asked (said as a +// failure); a TypeScript fetch is given the group's packages. +func (st *state) memberTarget(ctx context.Context, p Pull, bm string, group []buildstep.FetchPackage) *target { t := &target{pull: p, module: bm, language: st.j.languageOf(bm)} if tc, none := buildstep.AgreedToolchain(st.j.Toolchains.Entries(t.language)); none == "" { t.toolchain = tc.Version @@ -615,7 +632,7 @@ func (st *state) askMember(ctx context.Context, p Pull, bm string, group []build beside, err := st.j.besideFor(ctx, st.c, p) if err != nil { st.fail(forgeMachine(err, bm), err.Error()+unheardWords(err)) - return + return nil } if len(beside) > 0 { args.Beside = fetchBeside(beside) @@ -628,7 +645,7 @@ func (st *state) askMember(ctx context.Context, p Pull, bm string, group []build } t.fetchArgs = args st.targets = append(st.targets, t) - st.ask(ctx, t, merged) + return t } // groupPackages are the TypeScript packages the group's merges change (design 47 G A-284, A-285): every package of a @@ -664,20 +681,38 @@ func (st *state) failedSince(n int) bool { // need that is busy is waited for, and short enough that a mesh where no node has it says so. const DefaultDeclineBound = 30 * time.Minute -// ask puts the target's check ask, waits for the step of the node that pulled it and judges its answer: the fetch, then -// each level's digest, kept verdicts and tests. When the answer leaves packages unmet, -// their tests needing what that node lacks (issue 890), the passes are kept and the levels from the first such one are -// asked again of a node that has it (placed). -func (st *state) ask(ctx context.Context, t *target, merged *[]Fetched) { - st.askNeeding(ctx, t, merged, st.levelsOf(), nil, nil) -} - -// askNeeding is ask with the levels to run, what the node must have and the packages it tests: none for the first ask +// askNeeding puts the target's check ask, waits for it and judges its answer, with the levels to run, what the node must have and the packages it tests: none for the first ask // of a target; for the placed one the unmet packages' needs and those packages alone, so no package the first ask judged // is run or counted again (ADR 0552 §4). func (st *state) askNeeding(ctx context.Context, t *target, merged *[]Fetched, levels, needs, packages []string) { + q := st.prepareAsk(ctx, t, merged, levels, needs, packages) + if q == nil || !st.putAsk(ctx, q) { + return + } + q.wait(ctx, st) + st.judgeAsk(ctx, q) +} + +// asking is one check ask put and waited for: its target, the merges its fetch adds to, the ask as put last (a declined +// one is put again under a new id) and, once done is closed, the step it ended with or why it has none. +type asking struct { + t *target + merged *[]Fetched + a buildstep.StepAsk + kind string + needs []string + placed bool + step buildstep.Step + failed bool // the ask ended with no step to judge: failOn and failWords say why + failOn string + failWords string + done chan struct{} +} + +// prepareAsk makes the target's ask with the levels to run, what the node must have and the packages it tests; nil when +// it cannot be asked (said as a failure). +func (st *state) prepareAsk(ctx context.Context, t *target, merged *[]Fetched, levels, needs, packages []string) *asking { p := t.pull - placed := len(needs) > 0 job := &buildstep.CheckJob{Fetch: t.fetchArgs, Levels: levels, Kept: true, Packages: packages, Toolchain: t.agreed} if st.j.Nodes != nil { names, err := st.j.Nodes.Names(ctx) @@ -686,7 +721,7 @@ func (st *state) askNeeding(ctx context.Context, t *target, merged *[]Fetched, l } if err != nil { st.fail(t.module, fmt.Sprintf("the check of %s#%d is not asked: the nodes' names the names check needs cannot be read: %v", p.Repository, p.Number, err)) - return + return nil } job.NodeNames = names } @@ -705,38 +740,64 @@ func (st *state) askNeeding(ctx context.Context, t *target, merged *[]Fetched, l if a.Language == buildstep.AskGo { a.Packages = st.packagesOf(p.Repository) } - var step buildstep.Step + return &asking{t: t, merged: merged, a: a, kind: kind, needs: needs, placed: len(needs) > 0, done: make(chan struct{})} +} + +// putAsk puts the ask on its build seat's stream; false when it could not be put (said as a failure). +func (st *state) putAsk(ctx context.Context, q *asking) bool { + if err := st.put(ctx, q); err != nil { + st.fail(q.t.module, fmt.Sprintf("the check of %s#%d could not be asked: %v", q.t.pull.Repository, q.t.pull.Number, err)) + return false + } + return true +} + +// put puts the ask as it stands now. It touches only what is safe from a waiting goroutine: the progress, the case's +// ask ids and its watch. +func (st *state) put(ctx context.Context, q *asking) error { + p := q.t.pull + what := fmt.Sprintf("the %s ask %s of %s#%d", q.kind, q.a.ID, p.Repository, p.Number) + if q.placed { + what += " for a node with " + strings.Join(q.needs, ", ") + } + st.c.Progress.Set(what, "") + st.c.Asked(q.a.ID) + if err := st.j.Asks.Put(ctx, q.a); err != nil { + return err + } + if watch := st.c.Watch; watch != nil { + watch(q.a.ID, nil) // put: a try names it at once, not only when it ends (issue 1890) + } + return nil +} + +// wait waits for the step of the node that pulled the ask, putting a declined ask again for a node that has its need, +// and closes done. It runs on its own goroutine beside the phase's other asks, so it changes nothing of the check but +// the asking itself (the facts and failures are the judging's, in order). +func (q *asking) wait(ctx context.Context, st *state) { + defer close(q.done) + p := q.t.pull var firstDecline time.Time for { - what := fmt.Sprintf("the %s ask %s of %s#%d", kind, a.ID, p.Repository, p.Number) - if placed { - what += " for a node with " + strings.Join(needs, ", ") - } - st.c.Progress.Set(what, "") - st.c.Asked(a.ID) - if err := st.j.Asks.Put(ctx, a); err != nil { - st.fail(t.module, fmt.Sprintf("the check of %s#%d could not be asked: %v", p.Repository, p.Number, err)) - return - } waiting := ctx if watch := st.c.Watch; watch != nil { - id := a.ID - watch(id, nil) // put: a try names it at once, not only when it ends (issue 1890) + id := q.a.ID waiting = buildrun.WithStepSeen(ctx, func(s buildstep.Step) { watch(id, &s) }) } - var err error - step, err = st.j.Asks.Wait(waiting, a) + step, err := st.j.Asks.Wait(waiting, q.a) if err != nil { + q.failed, q.failOn = true, q.t.module if ctx.Err() != nil { - st.fail(t.module, ctx.Err().Error()) - return + q.failWords = ctx.Err().Error() + } else { + q.failWords = fmt.Sprintf("the check of %s#%d: %v", p.Repository, p.Number, err) } - st.fail(t.module, fmt.Sprintf("the check of %s#%d: %v", p.Repository, p.Number, err)) return } + q.step = step d, declined := buildstep.DeclinedOf(step) if !declined { - break + return } // A node that lacks the need dropped the ask (issue 890): it is put again, for a node that has it to take; the // node that declined looks at the stream again only after its next pull's wait. @@ -749,17 +810,35 @@ func (st *state) askNeeding(ctx context.Context, t *target, merged *[]Fetched, l bound = DefaultDeclineBound } if now.Sub(firstDecline) > bound { - lacks := strings.Join(needs, ", ") + lacks := strings.Join(q.needs, ", ") if lacks == "" { lacks = strings.Join(d.Lacks, ", ") // a node off the agreed toolchain declined it (ADR 0584) } - st.fail(step.Node, fmt.Sprintf("the check of %s#%d needs a node with %s, and none took it in %s: every node that pulled it lacks it (last %s)", - p.Repository, p.Number, lacks, bound, step.Node)) + q.failed, q.failOn = true, step.Node + q.failWords = fmt.Sprintf("the check of %s#%d needs a node with %s, and none took it in %s: every node that pulled it lacks it (last %s)", + p.Repository, p.Number, lacks, bound, step.Node) + return + } + st.j.logf("%s#%d: the ask %s was declined on %s, which lacks %s; asked again", p.Repository, p.Number, q.a.ID, step.Node, strings.Join(d.Lacks, ", ")) + q.a.ID, q.a.At = buildstep.NewID(), now.UTC() + if err := st.put(ctx, q); err != nil { + q.failed, q.failOn = true, q.t.module + q.failWords = fmt.Sprintf("the check of %s#%d could not be asked: %v", p.Repository, p.Number, err) return } - st.j.logf("%s#%d: the ask %s was declined on %s, which lacks %s; asked again", p.Repository, p.Number, a.ID, step.Node, strings.Join(d.Lacks, ", ")) - a.ID, a.At = buildstep.NewID(), now.UTC() } +} + +// judgeAsk judges the step an ask ended with: the fetch, then each level's digest, kept verdicts and tests. When the +// answer leaves packages unmet, their tests needing what that node lacks (issue 890), the passes are kept and the levels +// from the first such one are asked again of a node that has it (placed). +func (st *state) judgeAsk(ctx context.Context, q *asking) { + if q.failed { + st.fail(q.failOn, q.failWords) + return + } + t, merged, needs, placed, step := q.t, q.merged, q.needs, q.placed, q.step + p := t.pull // 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) {