diff --git a/go/buildstep/askrun.go b/go/buildstep/askrun.go index f0cef51..d189578 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. @@ -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 diff --git a/go/buildstep/askrun_test.go b/go/buildstep/askrun_test.go index 8ceb678..19bed04 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,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 diff --git a/go/buildstep/pull.go b/go/buildstep/pull.go index 80b41d8..960d02e 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, 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) 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/buscompose/buscompose_test.go b/go/buscompose/buscompose_test.go index f5244ba..e7ff21c 100644 --- a/go/buscompose/buscompose_test.go +++ b/go/buscompose/buscompose_test.go @@ -69,6 +69,37 @@ func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) { } }) } + // An asker on a seat pulled by kind hears what that seat's workers gave up on, and no other stream's dead letters + // (novox/hq issue 734: a check ask given up during a bus reload left its try running for ever). + t.Run("an asker reads its seat's dead letters", func(t *testing.T) { + build := Seat{Name: "node-build-go", Scope: "node", Accepts: []string{"delivery", "check", "recheck"}, PulledByKind: true} + perms, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "mesh-delivery-checker", Uses: []Seat{build}}) + if err != nil { + t.Fatal(err) + } + for subject, want := range map[string]bool{ + "$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1.mesh.events.dead.SEAT_NODE_BUILD_GO.>": true, + "$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1.mesh.events.dead.SEAT_NODE_BUILD_GO.w.next": true, + "$JS.API.CONSUMER.DELETE.DEAD_LETTERS.x1": true, + "$JS.API.CONSUMER.MSG.NEXT.DEAD_LETTERS.x1": true, + "$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1.mesh.events.dead.EVENTS.>": false, + "$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1": false, + "$JS.API.STREAM.PURGE.DEAD_LETTERS": false, + "$JS.FC.DEAD_LETTERS.x1": false, + } { + if got := MayPublish(perms, subject); got != want { + t.Errorf("may publish %s: %v, want %v", subject, got, want) + } + } + plain, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "shop", + Uses: []Seat{{Name: "telegram-sender", Accepts: []string{"send"}}}}) + if err != nil { + t.Fatal(err) + } + if MayPublish(plain, "$JS.API.CONSUMER.DELETE.DEAD_LETTERS.x1") { + t.Error("a user of a seat not pulled by kind reads DEAD_LETTERS") + } + }) t.Run("the controller's golden server configuration", func(t *testing.T) { seat := Seat{Name: "telegram-sender", Accepts: []string{"send"}, Emits: []string{"delivered", "failed"}} got, err := Compose( diff --git a/go/buscompose/seattraffic.go b/go/buscompose/seattraffic.go index 1b7095c..e89b208 100644 --- a/go/buscompose/seattraffic.go +++ b/go/buscompose/seattraffic.go @@ -24,8 +24,9 @@ import ( // subjects cover it, so what travels there — a code typed by the operator — is never persisted. A kinded // holder asks with its own kind; the modules that watch the seat answer. // -// And one read: **a holder's records**, a bucket the seat names, read by each user under its own name only -// (`$KV...>`), so an asker reads the state of its own asks and no other asker's. +// And two reads: **a holder's records**, a bucket the seat names, read by each user under its own name only +// (`$KV...>`), so an asker reads the state of its own asks and no other asker's; and on a seat pulled by +// kind, **what its workers gave up on**, the copies DEAD_LETTERS keeps of the seat's stream (DeadLetterReads, issue 734). // BuildAskMaxDeliver bounds the deliveries of a build seat's ask (issue 1907): mesh-sdk delivery.AskMaxDeliver, written // here as the broker mirrors the catalogue rather than importing it. @@ -129,6 +130,9 @@ func SeatTrafficOf(module string, holds, uses, watches []Seat) SeatTraffic { } t.Records = append(t.Records, "$JS.API.DIRECT.GET.KV_"+b+".$KV."+b+"."+module+".>") } + if s.PulledByKind && len(s.Accepts) > 0 && safeSubject.MatchString(s.Name) { + t.Records = append(t.Records, DeadLetterReads(seatStreamName(s.Name))...) + } } for _, w := range watches { if w.Kinded { @@ -151,6 +155,20 @@ func SeatTrafficOf(module string, holds, uses, watches []Seat) SeatTraffic { return t } +// DeadLetterReads is what a user of a seat pulled by kind is granted to hear which of the asks it put the seat's workers +// gave up on (novox/hq issue 734): an ordered consumer on DEAD_LETTERS filtered to the copies the controller keeps of that +// seat's stream (DeadLetterSubject(stream, *)), pulled, and its deletion. A copy carries the ask's own message id +// (Mesh-Dead-Msg-Id), so the asker ends the step it waits for instead of waiting for ever; it reads the seat's dead +// letters and no other stream's, and changes none. +func DeadLetterReads(stream string) []string { + return []string{ + "$JS.API.STREAM.INFO." + DeadLettersStream, + "$JS.API.CONSUMER.CREATE." + DeadLettersStream + ".*." + DeadLetterSubject(stream, ">"), + "$JS.API.CONSUMER.MSG.NEXT." + DeadLettersStream + ".*", + "$JS.API.CONSUMER.DELETE." + DeadLettersStream + ".*", + } +} + // seatTrafficGrants is the bus permissions seat traffic needs: the subjects themselves, and the JetStream API a // worker is pulled and acknowledged through and a record is read through. func seatTrafficGrants(t SeatTraffic) (pub, sub []string) { diff --git a/go/delivery/buildrun/asks_test.go b/go/delivery/buildrun/asks_test.go index 3365202..0f8540a 100644 --- a/go/delivery/buildrun/asks_test.go +++ b/go/delivery/buildrun/asks_test.go @@ -2,11 +2,15 @@ package buildrun import ( "context" + "encoding/json" + "errors" "strings" + "sync" "testing" "time" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/health" ) func goAsk(id string) delivery.StepAsk { @@ -48,6 +52,74 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) { if len(g.Asks()) != 0 { t.Fatal("a wait put an ask") } + // Issue 734: an ask the build seat gave up on (DEAD_LETTERS keeps its copy with the ask's id) ends its wait as failed + // with the words, though a node had taken it; under NotTakenBound one no node takes ends at the bound; a step written + // first is answered as the node wrote it. + dead := &DeadLetters{} + lw := LostAskWaiter{Inner: g, Dead: dead} + k = Asker{Publisher: g, Waiter: lw, Bound: time.Minute} + g.Write(delivery.Step{ID: "mvl0frv18d1970", Node: "ace", State: delivery.StepRunning, Started: time.Now()}) + given, ok := GivenUpOf(map[string][]string{"Mesh-Dead-Msg-Id": {"mvl0frv18d1970"}, "Mesh-Dead-Deliveries": {"5"}, + "Mesh-Dead-Stream": {"SEAT_NODE_BUILD_GO"}, "Mesh-Dead-Gave-Up": {"2026-10-26T12:26:40Z"}}) + if !ok { + t.Fatal("a dead letter's headers were not read") + } + go dead.Heard(given) + step, err = k.Wait(context.Background(), goAsk("mvl0frv18d1970")) + if err != nil || step.State != delivery.StepFailed || !IsLost(step) || + !strings.Contains(step.Reason, LostWords+" after 5 deliveries at 2026-10-26T12:26:40Z") { + t.Fatalf("a given-up ask: %+v, %v", step, err) + } + if _, err := k.Passed(step, nil); err == nil || !strings.Contains(err.Error(), "on "+LostNode+" failed: "+LostWords) { + t.Fatalf("a given-up ask's error: %v", err) + } + step, err = k.Wait(NotTakenBound(context.Background(), 20*time.Millisecond), goAsk("mvl0frv18d1971")) + if err != nil || !IsNotTaken(step) || IsLost(step) { + t.Fatalf("an ask no node took: %+v, %v", step, err) + } + g.Write(delivery.Step{ID: "mvl0frv18d1972", Node: "g14", State: delivery.StepPassed}) + if step, err = k.Wait(NotTakenBound(context.Background(), time.Minute), goAsk("mvl0frv18d1972")); err != nil || step.Node != "g14" { + t.Fatalf("a step written: %+v, %v", step, err) + } + // Review round 1 of issue 734: a hearing the bus refuses at start is deaf and says so, never holds the start up, and + // tries again on the next ask waited for; heard, it says so and hands the given-up asks on. + fol := &refusingFollower{refuse: 1} + var mu sync.Mutex + var told []Hearing + hearing := NewDeadLetterHearing(context.Background(), fol, []string{delivery.AskGo}, dead, nil, + func(h Hearing) { mu.Lock(); told = append(told, h); mu.Unlock() }) + hearing.Wait() + if why := hearing.Deaf(); !strings.Contains(why, "SEAT_NODE_BUILD_GO: permissions violation") { + t.Fatalf("a refused hearing says %q", why) + } + k = Asker{Publisher: g, Waiter: LostAskWaiter{Inner: g, Dead: dead, Hearing: hearing}, Bound: time.Minute} + g.Write(delivery.Step{ID: "mvl0frv18d1973", Node: "g14", State: delivery.StepPassed}) + if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1973")); err != nil { + t.Fatal(err) + } + hearing.Wait() + mu.Lock() + if hearing.Deaf() != "" || len(told) != 2 || told[0].Heard || !told[1].Heard || fol.calls() != 2 { + t.Fatalf("after an ask: deaf %q, told %+v, followed %d times", hearing.Deaf(), told, fol.calls()) + } + mu.Unlock() + fol.heard(GivenUp{AskID: "mvl0frv18d1974", Deliveries: 5}) + step, err = k.Wait(context.Background(), goAsk("mvl0frv18d1974")) + if err != nil || !IsLost(step) { + t.Fatalf("an ask heard given up after the hearing came back: %+v, %v", step, err) + } + for _, row := range []struct { + h *Hearing + want string + }{{nil, health.StateUnhealthy}, {&told[0], health.StateUnhealthy}, {&told[1], health.StateHealthy}} { + var raw []byte + if row.h != nil { + raw, _ = json.Marshal(row.h) + } + if a := HearingHealth(raw); a.State != row.want || a.Why == "" { + t.Errorf("health of %+v = %+v, want %s", row.h, a, row.want) + } + } } // Review (issue 734 path): an ask taken again by the same node (its build module replaced mid-run) is bounded from the @@ -69,3 +141,26 @@ func TestARedeliveryToTheSameNodeRestartsTheBound(t *testing.T) { t.Fatalf("step %+v, %v, want the redelivered run's pass", step, err) } } + +// refusingFollower refuses the first `refuse` follows, as the bus does before the asker's grant is composed. +type refusingFollower struct { + mu sync.Mutex + refuse int + n int + hand func(GivenUp) +} + +func (f *refusingFollower) Follow(_ context.Context, _ string, _ time.Time, heard func(GivenUp)) (func(), error) { + f.mu.Lock() + defer f.mu.Unlock() + f.n++ + if f.n <= f.refuse { + return nil, errors.New("permissions violation") + } + f.hand = heard + return func() {}, nil +} + +func (f *refusingFollower) calls() int { f.mu.Lock(); defer f.mu.Unlock(); return f.n } + +func (f *refusingFollower) heard(g GivenUp) { f.mu.Lock(); h := f.hand; f.mu.Unlock(); h(g) } diff --git a/go/delivery/buildrun/lostask.go b/go/delivery/buildrun/lostask.go new file mode 100644 index 0000000..90a9ece --- /dev/null +++ b/go/delivery/buildrun/lostask.go @@ -0,0 +1,495 @@ +package buildrun + +import ( + "context" + "encoding/json" + "fmt" + "strconv" + "strings" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/health" +) + +// **An ask the build seat gave up on ends its step** (novox/hq issue 734). A build seat's worker offers an ask +// delivery.AskMaxDeliver times; one handed out that often without an answer (a bus reload that drops every hand-out, a node +// that dies taking it each time) is given up by the server, and the controller keeps a copy of it in DEAD_LETTERS under +// mesh.events.dead.., with the ask's message id (its ask id), how often it was handed out and when it +// was given up. No build node writes a step for it, so an asker that waits for the step waits for ever: a try stays +// running and holds a try place, a check's verdict stays running and its pull request waits. +// +// An asker of the build seats (the checker, the builder) hears those copies (DeadLetterHearing: an ordered consumer on +// DEAD_LETTERS filtered to the build seats' streams, granted to a user of a seat pulled by kind, buscompose +// DeadLetterReads) into DeadLetters, and LostAskWaiter ends the waiting step at once as failed on the node LostNode, its +// reason saying LostWords; the asker decides what follows (the checker ends a try as error and asks a check again, the +// builder asks its part of a build again). +// +// Belt and braces, where the asker says so in the context (NotTakenBound; the checker's tries): an ask no build node took +// within a bound ends the same way, saying NotTakenWords. An ask waits in the stream as long as the backlog is otherwise +// (ADR 0525 §9): a backlog is no fault. + +// LostNode is the node a step ended for a lost ask names: no build node wrote it. +const LostNode = "bus" + +// The words of a step ended for its ask. +const ( + LostWords = "its ask was given up by the build seat" + NotTakenWords = "no build node took its ask" +) + +// The headers the controller keeps a given-up message's copy with in DEAD_LETTERS (mesh-controller internal/link +// deadletters.go): the original message id (an ask's is its ask id), how often it was handed out, when it was given up. +const ( + DeadHeaderMsgID = "Mesh-Dead-Msg-Id" + DeadHeaderDeliveries = "Mesh-Dead-Deliveries" + DeadHeaderGaveUp = "Mesh-Dead-Gave-Up" + DeadHeaderStream = "Mesh-Dead-Stream" +) + +// DeadLettersStream is the stream the controller keeps given-up messages in, and DeadLetterFilter the subjects of one seat +// stream's copies there (buscompose DeadLetterSubject). +const DeadLettersStream = "DEAD_LETTERS" + +// DeadLetterFilter is the subjects DEAD_LETTERS keeps a seat stream's given-up messages under. +func DeadLetterFilter(stream string) string { return "mesh.events.dead." + stream + ".>" } + +// SeatStream is the stream of a build seat's language (buscompose seatStreamName of node-build-). +func SeatStream(language string) string { return "SEAT_NODE_BUILD_" + strings.ToUpper(language) } + +// GivenUp is one ask a build seat gave up on, as DEAD_LETTERS keeps it. +type GivenUp struct { + AskID string + Stream string + Deliveries int + At time.Time +} + +// GivenUpOf reads a DEAD_LETTERS copy's headers; false when it names no message id. +func GivenUpOf(header map[string][]string) (GivenUp, bool) { + get := func(k string) string { + if v := header[k]; len(v) > 0 { + return strings.TrimSpace(v[0]) + } + return "" + } + g := GivenUp{AskID: get(DeadHeaderMsgID), Stream: get(DeadHeaderStream)} + if g.AskID == "" { + return GivenUp{}, false + } + g.Deliveries, _ = strconv.Atoi(get(DeadHeaderDeliveries)) + g.At, _ = time.Parse(time.RFC3339Nano, get(DeadHeaderGaveUp)) + return g, true +} + +// Words are what a step ended for the given-up ask says. +func (g GivenUp) Words() string { + at := "at an unknown time" + if !g.At.IsZero() { + at = "at " + g.At.UTC().Format(time.RFC3339) + } + return fmt.Sprintf("%s after %d deliveries %s (kept in %s); ask again", LostWords, g.Deliveries, at, DeadLettersStream) +} + +// keptGivenUp is how many given-up asks DeadLetters remembers for a waiter that starts after the copy was heard. +const keptGivenUp = 1024 + +// DeadLetters holds the given-up asks heard, and hands each to the waiter of its ask. Safe for concurrent use. +type DeadLetters struct { + mu sync.Mutex + heard map[string]GivenUp + order []string + watchers map[string][]chan GivenUp +} + +// Heard takes one given-up ask in. +func (d *DeadLetters) Heard(g GivenUp) { + d.mu.Lock() + if d.heard == nil { + d.heard, d.watchers = map[string]GivenUp{}, map[string][]chan GivenUp{} + } + if _, ok := d.heard[g.AskID]; !ok { + d.order = append(d.order, g.AskID) + if len(d.order) > keptGivenUp { + delete(d.heard, d.order[0]) + d.order = d.order[1:] + } + } + d.heard[g.AskID] = g + ws := d.watchers[g.AskID] + delete(d.watchers, g.AskID) + d.mu.Unlock() + for _, w := range ws { + w <- g // buffered, one each + } +} + +// watch answers a channel that gets the ask's given-up copy once (at once when it was heard already), and its cancel. +func (d *DeadLetters) watch(askID string) (<-chan GivenUp, func()) { + ch := make(chan GivenUp, 1) + d.mu.Lock() + defer d.mu.Unlock() + if d.heard == nil { + d.heard, d.watchers = map[string]GivenUp{}, map[string][]chan GivenUp{} + } + if g, ok := d.heard[askID]; ok { + ch <- g + return ch, func() {} + } + d.watchers[askID] = append(d.watchers[askID], ch) + return ch, func() { + d.mu.Lock() + defer d.mu.Unlock() + ws := d.watchers[askID] + for i, w := range ws { + if w == ch { + d.watchers[askID] = append(ws[:i], ws[i+1:]...) + break + } + } + if len(d.watchers[askID]) == 0 { + delete(d.watchers, askID) + } + } +} + +type notTakenKey struct{} + +// NotTakenBound says, in ctx, that an ask no build node took within bound ends its step (a try's, never a check's). +func NotTakenBound(ctx context.Context, bound time.Duration) context.Context { + return context.WithValue(ctx, notTakenKey{}, bound) +} + +func notTakenBoundOf(ctx context.Context) time.Duration { + d, _ := ctx.Value(notTakenKey{}).(time.Duration) + return d +} + +// LostAskWaiter is an AskWaiter that ends an ask's step when the build seat gave the ask up (DeadLetters), or, under +// NotTakenBound, when no build node took it within the bound: it hands the asker a failed step whose reason says why, on +// the node "bus". Every step the build module writes is handed on as it is. +type LostAskWaiter struct { + Inner AskWaiter + Dead *DeadLetters + // Hearing, when set, is asked to hear DEAD_LETTERS again each time an ask is waited for while it cannot: an ask put is + // the event that makes hearing matter. + Hearing interface { + Deaf() string + Again() + } + Now func() time.Time +} + +func (w LostAskWaiter) now() time.Time { + if w.Now != nil { + return w.Now() + } + return time.Now() +} + +// WatchAsk is AskWaiter's. +func (w LostAskWaiter) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) { + if w.Hearing != nil && w.Hearing.Deaf() != "" { + w.Hearing.Again() + } + inner, err := w.Inner.WatchAsk(ctx, module, askID) + if err != nil { + return nil, err + } + var dead <-chan GivenUp + cancel := func() {} + if w.Dead != nil { + dead, cancel = w.Dead.watch(askID) + } + // The not-taken bound is a context's (ADR 0437 §8: no timer), ended when a node takes the ask. + var notTaken <-chan struct{} + bound := notTakenBoundOf(ctx) + stopBound := func() {} + if bound > 0 { + var nt context.Context + nt, stopBound = context.WithTimeout(context.Background(), bound) + notTaken = nt.Done() + } + out := make(chan delivery.Step) + go func() { + defer close(out) + defer cancel() + defer func() { stopBound() }() + send := func(s delivery.Step) bool { + select { + case out <- s: + return true + case <-ctx.Done(): + return false + } + } + ended := func(reason string) delivery.Step { + at := w.now().UTC() + return delivery.Step{ID: askID, Node: LostNode, State: delivery.StepFailed, Reason: reason, Ended: at} + } + for { + select { + case <-ctx.Done(): + return + case s, ok := <-inner: + if !ok { + return + } + if notTaken != nil { + stopBound() // a node took it: its step is bounded by the asker from its start + notTaken = nil + } + if !send(s) { + return + } + case g := <-dead: + dead = nil + send(ended(g.Words())) + return + case <-notTaken: + send(ended(fmt.Sprintf("%s within %s of its put; it was lost or waits behind a backlog; ask again", NotTakenWords, bound))) + return + } + } + }() + return out, nil +} + +// IsLost says whether a step was ended by LostAskWaiter because the build seat gave its ask up (LostWords). +func IsLost(s delivery.Step) bool { return s.Node == LostNode && strings.Contains(s.Reason, LostWords) } + +// IsNotTaken says whether a step was ended by LostAskWaiter because no build node took its ask within the bound. +func IsNotTaken(s delivery.Step) bool { + return s.Node == LostNode && strings.Contains(s.Reason, NotTakenWords) +} + +// **The asker hears DEAD_LETTERS, tries again when it cannot, and says whether it hears** (issue 734). The read is +// granted with the asker's seat traffic (buscompose DeadLetterReads); an asker started before its grant was composed is +// refused, and an asker that only tried once would stay deaf until it starts again, silently. DeadLetterHearing tries in +// the background, never holding the start up, and again on an event (Again): the asker's reconnect to the bus, and every +// ask waited for while it is deaf (LostAskWaiter.Hearing); no timer. Each change of whether it hears is said (Said), so +// the module's health tool says it: a deaf asker is not healthy. + +// Hearing is what a DeadLetterHearing says: whether it hears every build seat stream it follows, and why not. +type Hearing struct { + Heard bool `json:"heard"` + Streams []string `json:"streams"` + Why string `json:"why,omitempty"` + At time.Time `json:"at"` +} + +// HearingKey is the key a module keeps its Hearing under, in its own state (the checker's and the builder's hearing). +const HearingKey = "dead-letters" + +// DeadLetterFollower starts following one seat stream's copies in DEAD_LETTERS from since, handing each given-up ask to +// heard; it answers the stop, or why it cannot. +type DeadLetterFollower interface { + Follow(ctx context.Context, stream string, since time.Time, heard func(GivenUp)) (func(), error) +} + +// JetStreamDeadLetters is the bus side of a DeadLetterFollower: an ordered consumer on DEAD_LETTERS filtered to the +// stream's copies, which resumes where it was after a reconnect and, reset before it heard any, starts at since again. +type JetStreamDeadLetters struct{ JS jetstream.JetStream } + +// Follow is DeadLetterFollower's. +func (j JetStreamDeadLetters) Follow(ctx context.Context, stream string, since time.Time, heard func(GivenUp)) (func(), error) { + cons, err := j.JS.OrderedConsumer(ctx, DeadLettersStream, jetstream.OrderedConsumerConfig{ + FilterSubjects: []string{DeadLetterFilter(stream)}, DeliverPolicy: jetstream.DeliverByStartTimePolicy, OptStartTime: &since}) + if err != nil { + return nil, err + } + cc, err := cons.Consume(func(m jetstream.Msg) { + if g, ok := GivenUpOf(m.Headers()); ok { + heard(g) + } + }) + if err != nil { + return nil, err + } + return cc.Stop, nil +} + +// DeadLetterHearing follows the build seats' given-up asks into Dead. Make it with NewDeadLetterHearing. +type DeadLetterHearing struct { + follower DeadLetterFollower + streams []string + dead *DeadLetters + logf func(string, ...any) + said func(Hearing) + ctx context.Context + since time.Time + + mu sync.Mutex + stops map[string]func() + why map[string]string + trying bool + again bool + told *Hearing + wg sync.WaitGroup +} + +// NewDeadLetterHearing starts hearing the given-up asks of each language's build seat stream, in the background. said +// is told each change of whether it hears (nil tells nothing); ctx ends it. +func NewDeadLetterHearing(ctx context.Context, follower DeadLetterFollower, languages []string, dead *DeadLetters, + logf func(string, ...any), said func(Hearing)) *DeadLetterHearing { + h := &DeadLetterHearing{follower: follower, dead: dead, logf: logf, said: said, ctx: ctx, since: time.Now(), + stops: map[string]func(){}, why: map[string]string{}} + for _, l := range languages { + s := SeatStream(l) + h.streams = append(h.streams, s) + h.why[s] = "not tried yet" + } + if h.logf == nil { + h.logf = func(string, ...any) {} + } + h.Again() + return h +} + +// Deaf says why a stream is not heard, or "" when every one is. +func (h *DeadLetterHearing) Deaf() string { + h.mu.Lock() + defer h.mu.Unlock() + return h.deafLocked() +} + +func (h *DeadLetterHearing) deafLocked() string { + var out []string + for _, s := range h.streams { + if why, deaf := h.why[s]; deaf { + out = append(out, s+": "+why) + } + } + if len(out) == 0 { + return "" + } + return DeadLettersStream + " is not heard, so an ask a build seat gives up on is waited for until its bound: " + + strings.Join(out, "; ") +} + +// Again tries, in the background, every stream not heard yet; a call while it tries makes it try once more after. +func (h *DeadLetterHearing) Again() { + h.mu.Lock() + if h.trying { + h.again = true + h.mu.Unlock() + return + } + h.trying = true + h.mu.Unlock() + h.wg.Add(1) + go func() { + defer h.wg.Done() + for { + h.try() + h.mu.Lock() + if !h.again || h.ctx.Err() != nil { + h.trying = false + h.mu.Unlock() + return + } + h.again = false + h.mu.Unlock() + } + }() +} + +// Wait returns when no try runs. +func (h *DeadLetterHearing) Wait() { h.wg.Wait() } + +// Stop ends every stream's following. +func (h *DeadLetterHearing) Stop() { + h.mu.Lock() + defer h.mu.Unlock() + for s, stop := range h.stops { + stop() + delete(h.stops, s) + } +} + +func (h *DeadLetterHearing) try() { + for _, s := range h.streams { + h.mu.Lock() + _, deaf := h.why[s] + h.mu.Unlock() + if !deaf || h.ctx.Err() != nil { + continue + } + stream := s + stop, err := h.follower.Follow(h.ctx, stream, h.since, func(g GivenUp) { + h.logf("the ask %s on %s was given up after %d deliveries at %s: the wait for its step ends", g.AskID, g.Stream, + g.Deliveries, g.At.Format(time.RFC3339)) + h.dead.Heard(g) + }) + h.mu.Lock() + if err != nil { + h.why[stream] = err.Error() + h.logf("the asks %s gives up on are not heard (%s of %s); tried again at the next reconnect or ask: %v", + stream, DeadLetterFilter(stream), DeadLettersStream, err) + } else { + delete(h.why, stream) + h.stops[stream] = stop + h.logf("the asks %s gives up on are heard from %s", stream, DeadLettersStream) + } + h.mu.Unlock() + } + h.tell() +} + +// tell says the hearing when it changed. +func (h *DeadLetterHearing) tell() { + h.mu.Lock() + why := h.deafLocked() + now := Hearing{Heard: why == "", Streams: append([]string(nil), h.streams...), Why: why, At: time.Now().UTC()} + same := h.told != nil && h.told.Heard == now.Heard && h.told.Why == now.Why + if !same { + h.told = &now + } + h.mu.Unlock() + if !same && h.said != nil { + h.said(now) + } +} + +// HearingState is the name of the state an asker keeps its Hearing in (`_hearing`, key HearingKey), which its +// health tool reads. +const HearingState = "hearing" + +// SayHearing is a DeadLetterHearing's said that keeps each Hearing in kv under HearingKey; nil kv keeps nothing and +// says so once. +func SayHearing(ctx context.Context, kv jetstream.KeyValue, logf func(string, ...any)) func(Hearing) { + if kv == nil { + if logf != nil { + logf("whether %s is heard is not kept: the state %s is not there, so the health tool cannot say it", DeadLettersStream, HearingState) + } + return nil + } + return func(h Hearing) { + data, err := json.Marshal(h) + if err == nil { + _, err = kv.Put(ctx, HearingKey, data) + } + if err != nil && logf != nil { + logf("whether %s is heard could not be kept: %v", DeadLettersStream, err) + } + } +} + +// HearingHealth is the health answer of an asker's kept Hearing (raw, empty when none is kept): healthy while it hears +// DEAD_LETTERS, unhealthy saying why while it does not or has not said. +func HearingHealth(raw []byte) health.Answer { + if len(raw) == 0 { + return health.Unhealthy("it has not said whether it hears " + DeadLettersStream + " (issue 734): it is starting, or its state " + HearingState + " is not there") + } + var h Hearing + if err := json.Unmarshal(raw, &h); err != nil { + return health.Unhealthy("what it keeps of hearing " + DeadLettersStream + " is not a hearing: " + err.Error()) + } + if !h.Heard { + return health.Unhealthy(h.Why + " (since " + h.At.Format(time.RFC3339) + "; tried again at its next reconnect or ask)") + } + return health.Healthy("it hears " + DeadLettersStream + " for " + strings.Join(h.Streams, ", ")) +} diff --git a/go/delivery/stepasks.go b/go/delivery/stepasks.go index fed56a8..cba0858 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 @@ -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 } diff --git a/go/delivery/stepasks_test.go b/go/delivery/stepasks_test.go index 7029fe9..2f10122 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,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),