diff --git a/go/buildstep/askrun.go b/go/buildstep/askrun.go index f0cef51..12a87e3 100644 --- a/go/buildstep/askrun.go +++ b/go/buildstep/askrun.go @@ -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. @@ -320,6 +323,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 diff --git a/go/buildstep/askrun_test.go b/go/buildstep/askrun_test.go index 8ceb678..37d3a76 100644 --- a/go/buildstep/askrun_test.go +++ b/go/buildstep/askrun_test.go @@ -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,27 @@ 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) + } + } } // ADR 0535 §2, §4: a merge that changes the build module's own code is judged by the candidate's runner: its digest says diff --git a/go/buildstep/pull.go b/go/buildstep/pull.go index cc95414..f2e0fa4 100644 --- a/go/buildstep/pull.go +++ b/go/buildstep/pull.go @@ -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) + 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) if err != nil { p.logf("the ask %s cannot start on %s: %v", ask.ID, p.Node, err) diff --git a/go/buildstep/pull_test.go b/go/buildstep/pull_test.go index 9269c45..902a460 100644 --- a/go/buildstep/pull_test.go +++ b/go/buildstep/pull_test.go @@ -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. diff --git a/go/delivery/stepasks.go b/go/delivery/stepasks.go index 9008be8..ad00703 100644 --- a/go/delivery/stepasks.go +++ b/go/delivery/stepasks.go @@ -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 @@ -290,6 +372,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) @@ -433,6 +522,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 +538,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 +557,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 +577,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 } diff --git a/go/delivery/stepasks_test.go b/go/delivery/stepasks_test.go index 7029fe9..19caa85 100644 --- a/go/delivery/stepasks_test.go +++ b/go/delivery/stepasks_test.go @@ -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,18 @@ 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.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),