novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

The checker puts every ask of a phase at once, then waits for them all

Browse Source 
judgeAll put one check ask and blocked in its wait before the next. Within each of its two phases (the modules that are
not TypeScript, then the TypeScript modules once groupPackages is read) every ask is now put first and waited for on its
own goroutine; a declined ask is put again there as before, and the steps are judged in the order the asks were put,
so the verdict reads as before whatever order they end in. The ask splits into prepareAsk, putAsk, wait and judgeAsk;
the placed ask of unmet packages stays one after the other.

Builds: design 47 G A-304, ADR 0525 §4

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-12 21:39:20 +01:00
parent f468aa75e8
commit 7b9904b1fe
2 changed files with 189 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-checker/internal/check/examiner_test.go
+53
View File
@@ -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))
modules/mesh-delivery-checker/internal/check/judge.go
+136 -57
View File
@@ -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) {