Merge pull request 'A check ask can name needs and packages, and a node that lacks a need declines it (issue 890, ADR 0552)' (#91) from feat/890-placement into main

This commit was merged in pull request #91.
This commit is contained in:
2026-10-26 13:40:35 +00:00
6 changed files with 277 additions and 7 deletions
+22
View File
@@ -34,6 +34,9 @@ type AskSteps struct {
// the module's sweep.
Release func(workspace string) error
Logf func(format string, args ...any)
// Has says whether this node has a need a package's tests declare (delivery.AskNeeds, issue 890); a package needing
// what it lacks is not tested here but named in the level's Unmet, for the asker to place. Nil has nothing.
Has func(need string) bool
}
// RunAsk runs ask's job with steps and answers a delivery.BuildAnswer or a delivery.CheckAnswer.
@@ -231,6 +234,9 @@ func (s AskSteps) level(ctx context.Context, job delivery.CheckJob, level string
if !keepRunner {
runner = "" // an installed module that ignores the runner judges the level as it judges any other (ADR 0535 §2)
}
if whole && len(job.Packages) > 0 {
todo, whole = job.Packages, false // a placed ask tests its packages alone, whatever the digest said (ADR 0552 §4)
}
for _, priority := range []bool{true, false} {
args := withRunner(map[string]any{"workspace": f.Workspace, "level": level})
if job.Run != "" {
@@ -307,9 +313,16 @@ func (s AskSteps) plan(ctx context.Context, job delivery.CheckJob, level string,
return nil, true, runner != "" // a digest that names nothing or not the siblings keys no verdict: the level runs whole
}
la.Digest = &d
only := map[string]bool{}
for _, p := range job.Packages {
only[p] = true
}
var todo []string
for _, pd := range d.Packages {
p := pd.Package
if len(only) > 0 && !only[p] {
continue // a placed ask's other packages were judged by the first ask (ADR 0552 §4)
}
if job.Kept && s.Kept != nil {
tests, found, err := s.Kept(ctx, level, pd.Digest)
if err != nil && s.Logf != nil {
@@ -320,6 +333,15 @@ func (s AskSteps) plan(ctx context.Context, job delivery.CheckJob, level string,
continue
}
}
if needs := d.NeedsOf(p); len(delivery.Lacks(needs, s.Has)) > 0 {
// Its tests would only skip here (issue 890): it is left for a node that has what they need.
la.Unmet = append(la.Unmet, delivery.PackageNeeds{Package: p, Needs: needs})
if s.Logf != nil {
s.Logf("the %s level of %s is left to a node with %s: this node lacks %s", level, p,
strings.Join(needs, ", "), strings.Join(delivery.Lacks(needs, s.Has), ", "))
}
continue
}
todo = append(todo, p)
}
return todo, false, false
+42 -2
View File
@@ -20,6 +20,7 @@ type recorder struct {
fetch map[string]any
tests map[string]bool // package or module directory -> passed
digests map[string]string
needs []delivery.PackageNeeds
}
func (r *recorder) steps() AskSteps {
@@ -35,8 +36,10 @@ func (r *recorder) steps() AskSteps {
Build: call("build", func(args map[string]any) any {
return map[string]any{"digest": "sha256:" + args["artifact"].(string), "reference": "artifact-store://" + args["artifact"].(string)}
}),
Copy: call("copy", func(args map[string]any) any { return map[string]any{"digest": "sha256:copied"} }),
Digest: call("digest", func(map[string]any) any { return map[string]any{"level": "unit", "packages": r.digests} }),
Copy: call("copy", func(args map[string]any) any { return map[string]any{"digest": "sha256:copied"} }),
Digest: call("digest", func(map[string]any) any {
return map[string]any{"level": "unit", "packages": r.digests, "needs": r.needs}
}),
Test: call("test", func(args map[string]any) any {
passed := true
if pkgs, ok := args["packages"].([]string); ok {
@@ -139,6 +142,43 @@ test {"level":"unit","packages":["x/pkg/a"],"workspace":"/w"}`
if u.Level != delivery.LevelUnit || len(u.Kept) != 1 || u.Kept[0].Package != "x/pkg/k" || u.Kept[0].Tests != 7 || len(u.Tests) != 2 || u.Digest == nil {
t.Fatalf("the unit level %+v", u)
}
// Issue 890/AC2: a package declaring node-build-node is not tested on a node that lacks it but named unmet, and is
// tested on a node that has it.
for _, has := range []bool{false, true} {
r.calls, r.tests["x/pkg/b"] = nil, true
r.needs = []delivery.PackageNeeds{{Package: "x/pkg/a", Needs: []string{delivery.NeedNodeBuildNode}}}
steps := r.steps()
if has {
steps.Has = func(n string) bool { return n == delivery.NeedNodeBuildNode }
}
out, err := RunAsk(context.Background(), steps, ask(delivery.CheckJob{Levels: []string{delivery.LevelUnit}, Kept: true,
Fetch: map[string]any{"repository": "novox/x", "commit": runCommit}}))
if err != nil {
t.Fatal(err)
}
u := out.(delivery.CheckAnswer).Levels[0]
tested := strings.Contains(strings.Join(r.calls, "\n"), `"x/pkg/a"`)
if tested != has || (len(u.Unmet) == 0) == !has || (!has && (u.Unmet[0].Package != "x/pkg/a" || delivery.UnmetNeeds([]delivery.LevelAnswer{u})[0] != delivery.NeedNodeBuildNode)) {
t.Fatalf("has %v: tested x/pkg/a %v, unmet %+v, calls %v", has, tested, u.Unmet, r.calls)
}
}
// A placed ask naming its packages tests those alone and takes no other's kept verdict, also when the level would
// run whole (ADR 0552 §4).
for _, digests := range []map[string]string{r.digests, nil} {
r.calls, r.digests = nil, digests
steps := r.steps()
steps.Has = func(string) bool { return true }
out, err := RunAsk(context.Background(), steps, ask(delivery.CheckJob{Levels: []string{delivery.LevelUnit}, Kept: true, Packages: []string{"x/pkg/a"},
Fetch: map[string]any{"repository": "novox/x", "commit": runCommit}}))
if err != nil {
t.Fatal(err)
}
u := out.(delivery.CheckAnswer).Levels[0]
if last := r.calls[len(r.calls)-1]; !strings.HasSuffix(last, `"packages":["x/pkg/a"],"workspace":"/w"}`) || len(u.Kept) != 0 || len(u.Tests) != 1 {
t.Fatalf("digests %v: calls %v, kept %+v: want x/pkg/a alone tested, nothing taken kept", digests != nil, r.calls, u.Kept)
}
}
}
// ADR 0535 §2, §4: a merge that changes the build module's own code is judged by the candidate's runner: its digest says
+55
View File
@@ -6,6 +6,7 @@ import (
"errors"
"fmt"
"sort"
"strings"
"sync"
"time"
@@ -76,6 +77,11 @@ type Puller struct {
// for this node, and wakes the puller on each change (watchCapacity): WatchConfigServed in a module, a test's own
// double. Until it hands a value the puller judges by Capacity.
WatchConfig func(key string, handle func(stdio.ProvisionChange) error) error
// Has says whether this node has a need an ask names (delivery.AskNeeds, issue 890). An ask needing what it lacks is
// declined: its step ends refused as declined (delivery.StepDeclined), the message is dropped for its asker to put
// again, and the worker it came from is not looked at again until the next pull's wait, so a node lacking the need
// does not take the asker's next copy at once. Nil has nothing.
Has func(need string) bool
mu sync.Mutex
inHand int
@@ -86,6 +92,9 @@ type Puller struct {
// watched is the capacity the config watch handed last, used in place of Capacity once hasWatched.
watched int
hasWatched bool
// declined are the kinds whose worker handed an ask this node declined, left out of the pulls without a wait until
// the next pull's wait ends.
declined map[string]bool
}
func (p *Puller) init() {
@@ -283,17 +292,59 @@ func (p *Puller) Pull(ctx context.Context) error {
// next is one ask from the workers in their order: one already waiting on delivery, check or recheck, and when none
// waits, a pull of the delivery worker that waits for one.
//
// A kind whose worker handed an ask this node declined is left out until that wait ended (issue 890): the pull's own
// wait, no timer of its own.
func (p *Puller) next(ctx context.Context, wait time.Duration) (*Pulled, error) {
p.mu.Lock()
declined := p.declined
p.mu.Unlock()
for _, kind := range delivery.AskKinds {
if declined[kind] {
continue
}
msg, err := p.Source.Next(ctx, kind, 0)
if err != nil || msg != nil {
return withKind(msg, kind), err
}
}
msg, err := p.Source.Next(ctx, delivery.AskDelivery, wait)
p.mu.Lock()
p.declined = nil
p.mu.Unlock()
return withKind(msg, delivery.AskDelivery), err
}
// decline ends the step of an ask this node lacks a need of, refused as declined with what it lacks, and drops the
// message: the asker puts the ask again under a new id (issue 890). A message whose decline cannot be written is handed
// back, as one whose start cannot be.
func (p *Puller) decline(msg *Pulled, ask delivery.StepAsk, lacks []string) {
step, key, err := begin(p.Put, p.Node, ask.Kind+" ask", ask.ID, AskWhat(ask))
if err == nil {
why := Refuse("the ask %s needs %s, and %s lacks %s: it is left to a node that has it", ask.ID,
strings.Join(ask.Needs, ", "), p.Node, strings.Join(lacks, ", "))
step.Ended = now().UTC()
step.State, step.Reason = delivery.StepRefused, why.Error()
step.Result, err = json.Marshal(delivery.StepDeclined{Declined: true, Needs: ask.Needs, Lacks: lacks})
if err == nil {
err = writeEnd(p.Put, key, step)
}
}
if err != nil {
p.logf("the ask %s cannot be declined on %s, so it is given back: %v", ask.ID, p.Node, err)
_ = msg.Nak()
return
}
p.logf("the ask %s is declined on %s: it needs %s, and this node lacks %s", ask.ID, p.Node, strings.Join(ask.Needs, ", "), strings.Join(lacks, ", "))
_ = msg.Term()
p.mu.Lock()
if p.declined == nil {
p.declined = map[string]bool{}
}
p.declined[msg.Kind] = true
p.mu.Unlock()
}
func withKind(msg *Pulled, kind string) *Pulled {
if msg != nil {
msg.Kind = kind
@@ -314,6 +365,10 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) {
_ = msg.Term()
return
}
if lacks := delivery.Lacks(ask.Needs, p.Has); len(lacks) > 0 {
p.decline(msg, ask, lacks)
return
}
step, key, err := begin(p.Put, p.Node, ask.Kind+" ask", ask.ID, AskWhat(ask))
if err != nil {
p.logf("the ask %s cannot start on %s: %v", ask.ID, p.Node, err)
+23 -4
View File
@@ -137,23 +137,42 @@ func TestPullHoldsAtMostItsCapacity(t *testing.T) {
waitFor(t, "all acked", func() bool { a, _, _ := src.Fates(); return len(a) == 3 })
}
// An ask that is not one, or not this seat's, is dropped and runs nowhere.
// An ask that is not one, or not this seat's, is dropped and runs nowhere. And issue 890/AC1: a check ask needing
// node-build-node is declined on a node that lacks it (its step refused as declined, naming what it lacks, the message
// dropped for the asker to put again), and runs on a node that has it.
func TestPullDropsWhatIsNoAsk(t *testing.T) {
src, st := NewFakeAsks(), newSteps()
src.Publish(delivery.AskDelivery, []byte(`{"v":1}`))
other := testAsk(delivery.AskDelivery)
other.Language = delivery.AskNode
_ = src.PublishAsk(other)
placed := testAsk(delivery.AskCheck)
placed.Needs = []string{delivery.NeedNodeBuildNode}
if err := src.PublishAsk(placed); err != nil {
t.Fatal(err)
}
ran := false
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 1 },
p := &Puller{Source: src, Language: delivery.AskGo, Node: "ace", Put: st.put, Capacity: func() int { return 1 },
PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { ran = true; return nil, nil }}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go p.Pull(ctx)
waitFor(t, "two termed", func() bool { _, tm, _ := src.Fates(); return len(tm) == 2 })
waitFor(t, "three termed", func() bool { _, tm, _ := src.Fates(); return len(tm) == 3 })
if ran {
t.Fatal("a wrong ask ran")
t.Fatal("a wrong or a declined ask ran")
}
key, _ := delivery.AskStepKey("ace", placed.ID)
s, _ := st.get(key)
if d, ok := delivery.DeclinedOf(s); !ok || len(d.Lacks) != 1 || d.Lacks[0] != delivery.NeedNodeBuildNode || !strings.Contains(s.Reason, "lacks node-build-node") {
t.Fatalf("the declined step = %+v", s)
}
has, hasSteps := NewFakeAsks(), newSteps()
_ = has.PublishAsk(placed)
q := &Puller{Source: has, Language: delivery.AskGo, Node: "g14", Put: hasSteps.put, Capacity: func() int { return 1 },
PullWait: 10 * time.Millisecond, Has: func(n string) bool { return n == delivery.NeedNodeBuildNode },
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }}
go q.Pull(ctx)
waitFor(t, "the placed ask acked where its need is met", func() bool { a, _, _ := has.Fates(); return len(a) == 1 })
}
// load reads the recorded steps state's raw values: the pause key is not a step, so it is kept apart.
+117 -1
View File
@@ -45,6 +45,86 @@ const (
// AskLanguages are the build seats' languages.
var AskLanguages = []string{AskGo, AskNode, AskImage}
// **An ask may name what its node needs** (issue 890, novox/hq ADR 0525 §1 as the placement record amends it). The seat's
// workers are shared, so the asker still chooses no node; a check ask whose packages' tests need something only some
// nodes have names it in Needs, and a holder whose node lacks one of them takes no part of the ask: it ends the ask's step
// refused as declined (StepDeclined) and drops the message, and the asker puts the ask again under a new id, so a node
// that has the need takes it. A need is a node seat whose holder installs what the tests read on its node; a package
// declares it beside its tests, and the build module's digest reports it (LevelDigest.Needs).
// NeedNodeBuildNode is the need of a package whose integration tests read the Node build module's installed toolchain
// (ADR 0533 §1): the node holds the seat node-build-node, so its toolchain is installed where the Go build module's
// integration level is given it.
const NeedNodeBuildNode = "node-build-node"
// AskNeeds are the needs an ask may name.
var AskNeeds = []string{NeedNodeBuildNode}
// CheckNeed refuses a need that is not one of AskNeeds.
func CheckNeed(need string) error {
for _, n := range AskNeeds {
if need == n {
return nil
}
}
return fmt.Errorf("delivery: %q is not a need an ask may name (%s)", need, strings.Join(AskNeeds, ", "))
}
// Lacks is what of needs has does not say the node has, in the order of needs; a nil has has nothing.
func Lacks(needs []string, has func(need string) bool) []string {
var out []string
for _, n := range needs {
if has == nil || !has(n) {
out = append(out, n)
}
}
return out
}
// PackageNeeds is one package and what its tests need of the node at a level.
type PackageNeeds struct {
Package string `json:"package"`
Needs []string `json:"needs"`
}
// StepDeclined is the result of the step of an ask a holder declined: the ask's needs, and those the holder's node lacks.
// The step's state is refused; the asker puts the ask again under a new id (DeclinedOf).
type StepDeclined struct {
Declined bool `json:"declined"`
Needs []string `json:"needs"`
Lacks []string `json:"lacks"`
}
// DeclinedOf says whether step is a declined ask's, and what its node lacked.
func DeclinedOf(step Step) (StepDeclined, bool) {
if step.State != StepRefused || len(step.Result) == 0 {
return StepDeclined{}, false
}
var d StepDeclined
if err := json.Unmarshal(step.Result, &d); err != nil || !d.Declined {
return StepDeclined{}, false
}
return d, true
}
// UnmetNeeds are the needs of the packages the levels left untested (LevelAnswer.Unmet), each once, sorted.
func UnmetNeeds(levels []LevelAnswer) []string {
seen := map[string]bool{}
var out []string
for _, l := range levels {
for _, u := range l.Unmet {
for _, n := range u.Needs {
if !seen[n] {
seen[n] = true
out = append(out, n)
}
}
}
}
sort.Strings(out)
return out
}
// AskRuns is how many times an ask is run before it is taken for poison (issue 1907): an ask whose run never ends (it
// takes its node down, or its process dies with it) is delivered again by its stream to the next free node (ADR 0525
// §6); the holder that takes it past AskRuns runs it no more, writes its step failed saying so, and ends it, so the
@@ -171,6 +251,8 @@ type StepAsk struct {
Build *BuildJob `json:"build,omitempty"`
Check *CheckJob `json:"check,omitempty"`
Compose *ComposeJob `json:"compose,omitempty"`
// Needs are what the node that runs the ask must have (AskNeeds); a check ask's alone. Empty: any node.
Needs []string `json:"needs,omitempty"`
}
// ComposeJob is a compose ask's job (novox/hq ADR 0482 clause 4, issue 1622): fetch the commit, build the composer
@@ -255,6 +337,9 @@ type CheckJob struct {
RunnerWhen string `json:"runner_when,omitempty"`
// Run is a try's test pattern (go test -run): each level runs it alone, digesting nothing.
Run string `json:"run,omitempty"`
// Packages, when set, are the only packages each level tests (issue 890, ADR 0552 §4): a placed ask names the
// packages the first ask left unmet, and the others are neither tested nor taken kept, so nothing is judged twice.
Packages []string `json:"packages,omitempty"`
}
func (a StepAsk) check() error {
@@ -290,6 +375,13 @@ func (a StepAsk) check() error {
return fmt.Errorf("delivery: the ask %s is a check on the delivery stream, which carries trunk builds alone", a.ID)
case a.Check != nil && a.Language == AskImage:
return fmt.Errorf("delivery: the ask %s is a check of an image, which has no tests", a.ID)
case len(a.Needs) > 0 && a.Check == nil:
return fmt.Errorf("delivery: the ask %s names needs and is no check, whose tests alone need a node", a.ID)
}
for _, n := range a.Needs {
if err := CheckNeed(n); err != nil {
return fmt.Errorf("the ask %s: %w", a.ID, err)
}
}
if a.Build != nil {
return a.Build.check(a.ID)
@@ -350,6 +442,11 @@ func (c CheckJob) check(id string) error {
return err
}
}
for _, p := range c.Packages {
if p == "" || strings.ContainsAny(p, " \t\n") {
return fmt.Errorf("delivery: the check ask %s names a package %q that is no import path", id, p)
}
}
return checkFetchArgs(id, c.Fetch)
}
@@ -433,6 +530,9 @@ type LevelAnswer struct {
Digest *LevelDigest `json:"digest,omitempty"`
Kept []KeptTaken `json:"kept,omitempty"`
Tests []json.RawMessage `json:"tests,omitempty"`
// Unmet are the packages the level did not test on this node, because their tests need what it lacks (issue 890): the
// asker keeps what passed and asks them again of a node that has it (StepAsk.Needs).
Unmet []PackageNeeds `json:"unmet,omitempty"`
}
// LevelDigest is the digest's answer as a level keeps it in the ask's step result: its packages a list of
@@ -446,6 +546,8 @@ type LevelDigest struct {
Beside []string `json:"beside,omitempty"`
WithoutTests []string `json:"without_tests,omitempty"`
JudgedBy string `json:"judged_by,omitempty"`
// Needs are the packages whose tests at this level need something of the node (issue 890), as they declare it.
Needs []PackageNeeds `json:"needs,omitempty"`
}
// PackageDigest is one package's digest at a level.
@@ -463,6 +565,7 @@ func (d *LevelDigest) UnmarshalJSON(raw []byte) error {
Beside []string `json:"beside"`
WithoutTests []string `json:"without_tests"`
JudgedBy string `json:"judged_by"`
Needs []PackageNeeds `json:"needs"`
}
if err := json.Unmarshal(raw, &v); err != nil {
return err
@@ -482,7 +585,20 @@ func (d *LevelDigest) UnmarshalJSON(raw []byte) error {
}
}
sort.Slice(pkgs, func(i, j int) bool { return pkgs[i].Package < pkgs[j].Package })
*d = LevelDigest{Level: v.Level, Packages: pkgs, Beside: v.Beside, WithoutTests: v.WithoutTests, JudgedBy: v.JudgedBy}
*d = LevelDigest{Level: v.Level, Packages: pkgs, Beside: v.Beside, WithoutTests: v.WithoutTests, JudgedBy: v.JudgedBy, Needs: v.Needs}
return nil
}
// NeedsOf is what the package's tests need of the node at the digest's level; nil when it declares nothing.
func (d *LevelDigest) NeedsOf(pkg string) []string {
if d == nil {
return nil
}
for _, n := range d.Needs {
if n.Package == pkg {
return n.Needs
}
}
return nil
}
+18
View File
@@ -56,6 +56,7 @@ func TestAskRefusesUnknownKindAndLanguage(t *testing.T) {
"recheck": func(a *StepAsk) { a.Kind = AskRecheck },
"commit": func(a *StepAsk) { a.Build.Fetch = map[string]any{"repository": "novox/x", "commit": "main"} },
"path": func(a *StepAsk) { a.Build.Path = "" },
"needs": func(a *StepAsk) { a.Needs = []string{NeedNodeBuildNode} }, // issue 890: a build needs no node
}
for name, change := range cases {
a := buildAsk()
@@ -80,6 +81,23 @@ func TestAskRefusesUnknownKindAndLanguage(t *testing.T) {
if _, err := EncodeStepAsk(c); err == nil {
t.Error("a check of an image encoded")
}
// Issue 890: a check names a known need and round-trips it; an unknown need is refused.
c = checkAsk()
c.Check.Packages = []string{"a b"}
if _, err := EncodeStepAsk(c); err == nil {
t.Error("a package that is no import path encoded")
}
c.Check.Packages = []string{"x/pkg/a"}
c.Needs = []string{"gpu"}
if _, err := EncodeStepAsk(c); err == nil {
t.Error("an unknown need encoded")
}
c.Needs = []string{NeedNodeBuildNode}
if raw, err := EncodeStepAsk(c); err != nil {
t.Errorf("a check needing %s: %v", NeedNodeBuildNode, err)
} else if back, err := DecodeStepAsk(raw); err != nil || len(back.Needs) != 1 || back.Needs[0] != NeedNodeBuildNode {
t.Errorf("the need did not round-trip: %+v %v", back.Needs, err)
}
raw, _ := EncodeStepAsk(buildAsk())
for _, bad := range []string{
strings.Replace(string(raw), `"kind":"delivery"`, `"kind":"queue"`, 1),