diff --git a/modules/mesh-delivery-checker/go.mod b/modules/mesh-delivery-checker/go.mod index 271e3b0..a517885 100644 --- a/modules/mesh-delivery-checker/go.mod +++ b/modules/mesh-delivery-checker/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-checker go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026111253-9c61918ac36f + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026130530-33395f68b309 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-checker/go.sum b/modules/mesh-delivery-checker/go.sum index 346bdcc..9d7f193 100644 --- a/modules/mesh-delivery-checker/go.sum +++ b/modules/mesh-delivery-checker/go.sum @@ -1,5 +1,11 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026111253-9c61918ac36f h1:9+TZpNq8Ifj1+qmbidE6l/7qkbua4yLdzRMrezNvf1A= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026111253-9c61918ac36f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026123512-bb713103371a h1:ZxKOtc8LusZEEpbbu41lYxJiaITN8lZCIcasA5kdgGQ= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026123512-bb713103371a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026130105-45564e69a648 h1:UUhWHYnaTJwgkgoYhQtJrm+r5S5UNAQ4DRMlt4x21Hg= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026130105-45564e69a648/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026130530-33395f68b309 h1:mKJToG8E6ZnZj9700mtfu6YUNNnxzSj0KTfZplQkUCo= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026130530-33395f68b309/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-checker/internal/check/examiner_test.go b/modules/mesh-delivery-checker/internal/check/examiner_test.go index 25c9c1e..c8f5b13 100644 --- a/modules/mesh-delivery-checker/internal/check/examiner_test.go +++ b/modules/mesh-delivery-checker/internal/check/examiner_test.go @@ -116,6 +116,9 @@ type bench struct { fetch Fetched digest map[string]Digested // by level test func(c stepCall) (any, error) + // declines is how many asks naming needs a node lacking them declines before shanks, which has every need, takes + // one (issue 890); an ask naming none runs on shanks, which then has none. + declines int } // refsOf reads the siblings of a fetch's arguments, as they arrive in an ask (JSON) or as the checker gave them. @@ -151,8 +154,17 @@ func (b *bench) answer(a delivery.StepAsk) delivery.Step { 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) + if len(a.Needs) > 0 { + if b.declines > 0 { + b.declines-- + result, _ := json.Marshal(delivery.StepDeclined{Declined: true, Needs: a.Needs, Lacks: a.Needs}) + return delivery.Step{ID: a.ID, Tool: a.Kind + " ask", Node: "ace", State: delivery.StepRefused, Started: began, Ended: began, + Reason: "declined", Result: result} + } + steps.Has = func(string) bool { return true } + } + out, err := buildstep.RunAsk(context.Background(), steps, a) 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() @@ -276,6 +288,49 @@ func TestOnlyThePackageWithoutAKeptVerdictRuns(t *testing.T) { // 1288/AC1 if u := out.Verdict.Levels[delivery.LevelUnit]; u.Run != 2 || u.Kept != 40 { t.Errorf("unit = %+v, want 2 run and 40 kept", u) } + + // Issue 890/AC5: a package needing node-build-node is left unmet where the node lacks it; the check keeps the passes + // and asks the integration level again for a node with it, puts it again when a node declines it, and that node tests + // the package alone, its kept passes not counted twice. + // fixture/outside has no kept verdict after its pass (it reads outside itself: its digest moves every call), so a + // placed ask that ran more than its unmet packages would run and count it again. + b = newBench(t) + b.declines = 1 + outside := 0 + b.steps.handlers["build_go_digest"] = func(c stepCall) (any, error) { + d := echoBeside(b, b.digest[c.args["level"].(string)]) + if d.Level == "integration" { + outside++ + d.Packages = map[string]string{"fixture/slow": digB, "fixture/node": digC, "fixture/outside": fmt.Sprintf("%064x", outside)} + } + return d, nil + } + b.digest[delivery.LevelIntegration] = Digested{Level: "integration", + Needs: []delivery.PackageNeeds{{Package: "fixture/node", Needs: []string{delivery.NeedNodeBuildNode}}}} + out = b.judge(t, caseOf(t, pullOf(repoA, 7, head1))) + asks := b.asks.Asks() + if out.Verdict.State != delivery.VerdictPass || len(asks) != 3 || len(asks[0].Needs) != 0 { + t.Fatalf("verdict %+v, asks %+v: want a pass of three asks, the first naming no need", out.Verdict, asks) + } + for _, a := range asks[1:] { + if strings.Join(a.Needs, ",") != delivery.NeedNodeBuildNode || strings.Join(a.Check.Levels, ",") != delivery.LevelIntegration || strings.Join(a.Check.Packages, ",") != "fixture/node" { + t.Errorf("a placed ask %+v, want the integration level needing node-build-node", a) + } + } + tests = b.steps.tool("test") + if last := tests[len(tests)-1]; len(tests) != 3 || strings.Join(packagesOf(last), ",") != "fixture/node" || last.args["level"] != "integration" { + t.Errorf("test calls = %+v, want the unit level, the integration level of fixture/slow and fixture/outside, then fixture/node alone", tests) + } + slow := captured[Tested](t, "go-test-slow") + if i := out.Verdict.Levels[delivery.LevelIntegration]; i.Run != 2*slow.Tests || i.Kept != 0 { + t.Errorf("integration = %+v, want %d run and nothing counted kept", i, 2*slow.Tests) + } + // And where no node has the need, the check fails once the declines outlast the bound, naming the need. + b.kept, b.declines = newMemKV(), 1000 + b.ex.Kept, b.ex.DeclineBound = Kept{Store: b.kept}, 200*time.Millisecond + if v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict; v.State != delivery.VerdictFail || !strings.Contains(fmt.Sprint(v), "needs a node with node-build-node") { + t.Errorf("with no node having the need: %+v", v) + } } func TestAReAskRunsOnlyWhatFailedAndWhatChanged(t *testing.T) { // 1288/AC2, AC3 @@ -620,9 +675,8 @@ func TestNothingIsKeptForAWorkspaceGivenSiblingsWhenTheDigestDoesNotSayItCoversT // Beside): every check now places the change's own repository at its trunk (issue 1887 A12). func echoBeside(b *bench, d Digested) Digested { var lines []string - for _, f := range b.steps.tool("fetch") { - refs, _ := refsOf(f.args["beside"]), true - for _, r := range refs { + if fetches := b.steps.tool("fetch"); len(fetches) > 0 { // the workspace digested is the last fetch's (issue 890: a placed ask fetches again) + for _, r := range refsOf(fetches[len(fetches)-1].args["beside"]) { lines = append(lines, r.Repository+" "+r.Commit) } } diff --git a/modules/mesh-delivery-checker/internal/check/judge.go b/modules/mesh-delivery-checker/internal/check/judge.go index b136728..06f8f11 100644 --- a/modules/mesh-delivery-checker/internal/check/judge.go +++ b/modules/mesh-delivery-checker/internal/check/judge.go @@ -80,6 +80,9 @@ var askLanguage = map[string]string{buildrun.GoBuildModule: delivery.AskGo, buil // stopping at the first failure; it judges the step's answer as it judged the tools' answers. type Examiner struct { Asks Asks + // DeclineBound is how long a placed ask may go on being declined, from its first decline, before the check fails + // saying no node with its need took it (issue 890, ADR 0552); DefaultDeclineBound when zero. + DeclineBound time.Duration // Toolchains are the nodes' toolchains: a kept verdict names the toolchain the nodes agree on. Toolchains ToolchainEntries Kept Kept @@ -381,11 +384,25 @@ func (st *state) judgeAll(ctx context.Context) *delivery.Verdict { return nil } +// DefaultDeclineBound is an Examiner's DeclineBound when it names none: longer than a check runs, so a node with the +// 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, the -// pins of its go.mod files, then each level's digest, kept verdicts and tests. +// pins of its go.mod files, 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) *delivery.Verdict { + return 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 +// 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) *delivery.Verdict { p := t.pull - job := &delivery.CheckJob{Fetch: t.fetchArgs, Levels: st.levelsOf(), Kept: true} + placed := len(needs) > 0 + job := &delivery.CheckJob{Fetch: t.fetchArgs, Levels: levels, Kept: true, Packages: packages} if t.runner != "" { job.Runner, job.RunnerWhen = t.runner, buildGoDir } @@ -397,17 +414,46 @@ func (st *state) ask(ctx context.Context, t *target, merged *[]Fetched) *deliver 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()) + At: st.j.now().UTC(), Check: job, Needs: needs} + var step delivery.Step + 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, ", ") } - return fail(t.module, fmt.Sprintf("the check of %s#%d: %v", p.Repository, p.Number, err)) + st.c.Progress.Set(what, "") + 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)) + } + var err error + 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)) + } + d, declined := delivery.DeclinedOf(step) + if !declined { + break + } + // 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. + now := st.j.now() + if firstDecline.IsZero() { + firstDecline = now + } + bound := st.j.DeclineBound + if bound <= 0 { + bound = DefaultDeclineBound + } + if now.Sub(firstDecline) > bound { + return 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, strings.Join(needs, ", "), bound, step.Node)) + } + 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() } // The fetch and the digests are what of the ask's time its levels' running did not take (Took.Clone). took := time.Duration(0) @@ -437,12 +483,22 @@ func (st *state) ask(ctx context.Context, t *target, merged *[]Fetched) *deliver 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 + if !placed { // the placed ask fetched the same merge: its preview and pins were read from the first + *merged = append(*merged, t.fetched) + if v := st.checkPinsOf(ctx, t); v != nil { + return v + } } ran := 0.0 for _, la := range answer.Levels { + if placed { + if len(la.Unmet) > 0 { + st.clone += max(0, took-time.Duration(ran*float64(time.Second))) + return fail(step.Node, fmt.Sprintf("the %s level of %s#%d left %s untested on %s, which was asked for a node with %s", + la.Level, p.Repository, p.Number, unmetNames(la.Unmet), step.Node, strings.Join(needs, ", "))) + } + la.Kept = nil // it names its packages and takes no other's kept verdict; this is only a guard against an older module + } v, elapsed := st.level(ctx, t, la) ran += elapsed if v != nil { @@ -451,9 +507,48 @@ func (st *state) ask(ctx context.Context, t *target, merged *[]Fetched) *deliver } } st.clone += max(0, took-time.Duration(ran*float64(time.Second))) + if unmet := delivery.UnmetNeeds(answer.Levels); len(unmet) > 0 && !placed { + // Every level passed, its passes kept, and some packages were left to a node with what their tests need: the levels + // from the first that left one are asked of such a node, which takes the kept passes and tests those (issue 890). + var again, packages []string + seen := map[string]bool{} + for _, la := range answer.Levels { + if len(again) > 0 || len(la.Unmet) > 0 { + again = append(again, la.Level) + } + for _, u := range la.Unmet { + if !seen[u.Package] { + seen[u.Package] = true + packages = append(packages, u.Package) + } + } + } + st.j.logf("%s#%d: %s on %s left untested; asked of a node with %s", p.Repository, p.Number, unmetOf(answer.Levels), step.Node, strings.Join(unmet, ", ")) + return st.askNeeding(ctx, t, merged, again, unmet, packages) + } return nil } +// unmetNames names the unmet packages and their needs. +func unmetNames(unmet []delivery.PackageNeeds) string { + var out []string + for _, u := range unmet { + out = append(out, u.Package+" (needs "+strings.Join(u.Needs, ", ")+")") + } + return strings.Join(out, ", ") +} + +// unmetOf names the unmet packages of every level. +func unmetOf(levels []delivery.LevelAnswer) string { + var out []string + for _, la := range levels { + if len(la.Unmet) > 0 { + out = append(out, "at "+la.Level+" "+unmetNames(la.Unmet)) + } + } + return strings.Join(out, "; ") +} + // 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 every test result is read and counted, also after // one that did not pass, so the verdict names every failed test of the level (issue 1921). It answers the level's diff --git a/modules/mesh-delivery-checker/internal/check/results.go b/modules/mesh-delivery-checker/internal/check/results.go index 87b8e37..bd61b15 100644 --- a/modules/mesh-delivery-checker/internal/check/results.go +++ b/modules/mesh-delivery-checker/internal/check/results.go @@ -3,6 +3,8 @@ package check import ( "encoding/json" "strings" + + "git.novox.be/novox/mesh-sdk/go/delivery" ) // The shapes of what the build modules' steps answer, as their code writes them (mesh-catalog modules/mesh-build-go @@ -109,4 +111,6 @@ type Digested struct { WithoutTests []string `json:"without_tests,omitempty"` // JudgedBy is the program that ran the digest, as Tested.JudgedBy says. JudgedBy string `json:"judged_by,omitempty"` + // Needs are the packages whose tests need something of the node at this level (issue 890). + Needs []delivery.PackageNeeds `json:"needs,omitempty"` }