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

Merge pull request 'The checker asks the packages a node could not test again, on a node that has what they need (issue 890, ADR 0552)' (#72) from feat/890-placement into main

Browse Source
This commit was merged in pull request #72.
This commit is contained in:
mesh-tools
2026-10-26 13:40:45 +00:00
parent b3a4b09f5b 3073495a06
commit 1cbae7d41c
5 changed files with 179 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-checker/go.mod
+1 -1
View File
@@ -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
)
modules/mesh-delivery-checker/go.sum
+6
View File
@@ -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=
modules/mesh-delivery-checker/internal/check/examiner_test.go
+58 -4
View File
@@ -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)
}
}
modules/mesh-delivery-checker/internal/check/judge.go
+110 -15
View File
@@ -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
modules/mesh-delivery-checker/internal/check/results.go
+4
View File
@@ -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"`
}