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

mesh-delivery-checker: put every member's check ask of a group at once and judge the answers in member order, so a group's members run side by side (issue 1921, try speed S8)

Browse Source
This commit is contained in:
jschoubben
2026-10-25 01:15:28 +02:00
parent 6fe7b7dd6c
commit 4ccd78c9ec
2 changed files with 140 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-checker/internal/check/ask_all_test.go
+61
View File
@@ -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)
}
}
modules/mesh-delivery-checker/internal/check/judge.go
+79 -37
View File
@@ -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())